Implement remaining cleanup items prior to the next release

.dockerignore

@@ -1,7 +1,8 @@
 .git
 .gitignore
 **/*.md
+!docs/*.md
+!docs/**/*.md
 dist/
 tmp/
 .DS_Store
-

README.md

@@ -29,8 +29,10 @@ current working directory.
 - [Operations guide](docs/operations.md)
 - [Troubleshooting guide](docs/troubleshooting.md)
 - [Example configs](examples/)
+- [Go consumer guide](docs/consumers/api.md)
 - [Event wire contract](docs/integrations/events.md)
 - [Postgres table contract](docs/integrations/postgres.md)
+- [Feedkit integration notes](docs/integrations/feedkit.md)
 - [NWS integration notes](docs/integrations/nws.md)
 - [SPC integration notes](docs/integrations/spc.md)
 - [Open-Meteo integration notes](docs/integrations/openmeteo.md)

cmd/weatherfeeder/main_test.go

@@ -20,6 +20,7 @@ import (
 
 	wfnormalizers "gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers"
 	wfsources "gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
 )
 
 type testInput struct {
@@ -36,10 +37,10 @@ type testKindsSource struct {
 func (s testKindsSource) Kinds() []fkevent.Kind { return s.kinds }
 
 func TestValidateSourceExpectedKindsSubsetAllowed(t *testing.T) {
-	sc := config.SourceConfig{Kinds: []string{"observation"}}
+	sc := config.SourceConfig{Kinds: []string{standards.KindObservation}}
 	in := testKindsSource{
 		testInput: testInput{name: "test"},
-		kinds:     []fkevent.Kind{"observation", "forecast"},
+		kinds:     []fkevent.Kind{fkevent.Kind(standards.KindObservation), fkevent.Kind(standards.KindForecast)},
 	}
 
 	if err := fksources.ValidateExpectedKinds(sc, in); err != nil {
@@ -48,10 +49,10 @@ func TestValidateSourceExpectedKindsSubsetAllowed(t *testing.T) {
 }
 
 func TestValidateSourceExpectedKindsMismatchFails(t *testing.T) {
-	sc := config.SourceConfig{Kinds: []string{"alert"}}
+	sc := config.SourceConfig{Kinds: []string{standards.KindAlert}}
 	in := testKindsSource{
 		testInput: testInput{name: "test"},
-		kinds:     []fkevent.Kind{"observation", "forecast"},
+		kinds:     []fkevent.Kind{fkevent.Kind(standards.KindObservation), fkevent.Kind(standards.KindForecast)},
 	}
 
 	err := fksources.ValidateExpectedKinds(sc, in)
@@ -64,7 +65,7 @@ func TestValidateSourceExpectedKindsMismatchFails(t *testing.T) {
 }
 
 func TestValidateSourceExpectedKindsNoMetadataSkipsCheck(t *testing.T) {
-	sc := config.SourceConfig{Kinds: []string{"alert"}}
+	sc := config.SourceConfig{Kinds: []string{standards.KindAlert}}
 	in := testInput{name: "test"}
 
 	if err := fksources.ValidateExpectedKinds(sc, in); err != nil {
@@ -111,6 +112,10 @@ func TestMaintainedConfigExamplesLoad(t *testing.T) {
 func assertConfigSourcesBuildSchedulerJobs(t *testing.T, cfg *config.Config) {
 	t.Helper()
 
+	if len(cfg.Sources) == 0 {
+		t.Fatalf("config has no sources")
+	}
+
 	reg := fksources.NewRegistry()
 	wfsources.RegisterBuiltins(reg)
 
@@ -158,7 +163,7 @@ func TestNormalizeNoMatchPassThrough(t *testing.T) {
 	pl := &fkpipeline.Pipeline{Processors: chain}
 	in := fkevent.Event{
 		ID:        "evt-no-match",
-		Kind:      fkevent.Kind("observation"),
+		Kind:      fkevent.Kind(standards.KindObservation),
 		Source:    "test",
 		EmittedAt: time.Now().UTC(),
 		Schema:    "raw.weatherfeeder.unknown.v1",
@@ -188,7 +193,7 @@ func TestDedupeDropsSecondEventWithSameID(t *testing.T) {
 	pl := &fkpipeline.Pipeline{Processors: chain}
 	in := fkevent.Event{
 		ID:        "evt-dedupe-1",
-		Kind:      fkevent.Kind("observation"),
+		Kind:      fkevent.Kind(standards.KindObservation),
 		Source:    "test",
 		EmittedAt: time.Now().UTC(),
 		Schema:    "raw.weatherfeeder.unknown.v1",

docs/config.md

@@ -147,7 +147,7 @@ URL omits it or sets another unit system.
 
 ## SPC Convective Outlook Params
 
-`spc_convective_outlook` fetches the twelve required Day 1-3 GeoJSON outlook
+`spc_convective_outlook` fetches the nine required Day 1-3 GeoJSON outlook
 products and the three required Day 1-3 print pages as one atomic bundle.
 
 | Param | Required | Description |
@@ -165,8 +165,8 @@ products and the three required Day 1-3 print pages as one atomic bundle.
 
 GeoJSON product keys are `day1_categorical`, `day1_tornado`, `day1_hail`,
 `day1_wind`, `day2_categorical`, `day2_tornado`, `day2_hail`, `day2_wind`,
-`day3_categorical`, `day3_tornado`, `day3_hail`, and `day3_wind`.
+and `day3_categorical`. SPC does not provide Day 3 tornado, hail, or wind
-Discussion keys are `day1`, `day2`, and `day3`.
+GeoJSON products. Discussion keys are `day1`, `day2`, and `day3`.
 
 ```yaml
 sources:

docs/consumers/api.md

@@ -0,0 +1,92 @@
+# Consumer API Guide
+
+## Purpose
+
+This guide is for developers and LLM coding agents integrating `weatherfeeder`
+from another Go codebase.
+
+`weatherfeeder` is primarily a daemon, not an SDK. Its public integration
+surface is intentionally narrow:
+
+- `model`: canonical weather payload structs.
+- `standards`: schema strings, event kind strings, and shared WMO constants.
+- JSON event output from stdout and NATS sinks.
+- Postgres tables written by the optional Postgres sink.
+
+Packages under `internal/` are implementation details and are not public
+integration surfaces.
+
+## Recommended Workflow
+
+Consumers should switch on the event `schema` value and decode `payload` into
+the matching `model` type.
+
+Minimal example:
+
+```go
+package consumer
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"gitea.maximumdirect.net/ejr/weatherfeeder/model"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
+)
+
+type Event struct {
+	ID      string          `json:"id"`
+	Kind    string          `json:"kind"`
+	Schema  string          `json:"schema"`
+	Payload json.RawMessage `json:"payload"`
+}
+
+func Decode(payload []byte) (any, error) {
+	var evt Event
+	if err := json.Unmarshal(payload, &evt); err != nil {
+		return nil, err
+	}
+
+	switch evt.Schema {
+	case standards.SchemaWeatherObservationV1:
+		var out model.WeatherObservation
+		return &out, json.Unmarshal(evt.Payload, &out)
+	case standards.SchemaWeatherForecastV1:
+		var out model.WeatherForecastRun
+		return &out, json.Unmarshal(evt.Payload, &out)
+	case standards.SchemaWeatherForecastDiscussionV1:
+		var out model.WeatherForecastDiscussion
+		return &out, json.Unmarshal(evt.Payload, &out)
+	case standards.SchemaWeatherStoryV1:
+		var out model.WeatherStoryRun
+		return &out, json.Unmarshal(evt.Payload, &out)
+	case standards.SchemaWeatherAlertV1:
+		var out model.WeatherAlertRun
+		return &out, json.Unmarshal(evt.Payload, &out)
+	case standards.SchemaWeatherOutlookV1:
+		var out model.WeatherOutlookRun
+		return &out, json.Unmarshal(evt.Payload, &out)
+	default:
+		return nil, fmt.Errorf("unsupported weatherfeeder schema %q", evt.Schema)
+	}
+}
+```
+
+## Consumer Responsibilities
+
+- Treat event IDs as opaque.
+- Treat absent `omitempty` fields as unknown, not zero.
+- Prefer schema constants from `standards` over string literals in Go code.
+- Expect canonical numeric measurements to use metric units.
+- Expect canonical timestamps from normalizers to be UTC unless a field-specific
+  contract says otherwise.
+- Handle additive fields within the same schema version.
+- Do not import `internal/...` packages.
+
+## Canonical References
+
+- Public payload package: [`pkg-model.md`](pkg-model.md).
+- Public constants package: [`pkg-standards.md`](pkg-standards.md).
+- JSON event wire contract: [`../integrations/events.md`](../integrations/events.md).
+- Postgres table contract: [`../integrations/postgres.md`](../integrations/postgres.md).
+- Runtime and adapter architecture: [`../policy/architecture.md`](../policy/architecture.md).

docs/consumers/pkg-model.md

@@ -0,0 +1,63 @@
+# Package `model`
+
+## Import Path
+
+```go
+import "gitea.maximumdirect.net/ejr/weatherfeeder/model"
+```
+
+## Purpose
+
+Package `model` defines `weatherfeeder`'s canonical weather payload structs.
+These structs are emitted as the `payload` of canonical `weather.*.v1` events
+and are also the domain types consumed by downstream applications such as
+`weatherapi`.
+
+The JSON field tags on these structs are part of the wire contract. For the full
+field-by-field JSON contract, use the [event wire contract](../integrations/events.md).
+
+## Payload Types
+
+Current canonical schema families map to these public types:
+
+| Schema | Primary type |
+|---|---|
+| `weather.observation.v1` | `WeatherObservation` |
+| `weather.forecast.v1` | `WeatherForecastRun` |
+| `weather.forecast_discussion.v1` | `WeatherForecastDiscussion` |
+| `weather.weather_story.v1` | `WeatherStoryRun` |
+| `weather.alert.v1` | `WeatherAlertRun` |
+| `weather.outlook.v1` | `WeatherOutlookRun` |
+
+Related child types include:
+
+- `WeatherObservationPresentWeather`
+- `WeatherForecastPeriod`
+- `WeatherForecastDiscussionSection`
+- `WeatherStory`
+- `WeatherAlert`
+- `WeatherAlertReference`
+- `WeatherOutlook`
+- `WMOCode`
+
+## Wire And Compatibility Rules
+
+- JSON tags define canonical payload field names.
+- Pointer fields and fields tagged `omitempty` are optional on the wire.
+- Missing optional fields mean unknown or not applicable.
+- Canonical measurements use metric units.
+- Canonical timestamps are `time.Time` values encoded by Go's JSON encoder.
+- Normalized canonical timestamps are UTC unless a field-specific contract says
+  otherwise.
+- Additive fields are compatible within a schema version.
+- Removing, renaming, or changing the meaning of a field requires a new schema
+  identifier.
+
+## Boundaries
+
+`model` should not depend on source adapters, sinks, SQL column names, provider
+HTTP shapes, or runtime configuration.
+
+Consumers should not rely on packages under `internal/...`. Use `model` with
+schema constants from [`standards`](pkg-standards.md) and the JSON contract in
+[`docs/integrations/events.md`](../integrations/events.md).

docs/consumers/pkg-standards.md

@@ -0,0 +1,90 @@
+# Package `standards`
+
+## Import Path
+
+```go
+import "gitea.maximumdirect.net/ejr/weatherfeeder/standards"
+```
+
+## Purpose
+
+Package `standards` defines stable identifiers and shared weather constants used
+by `weatherfeeder` producers and Go consumers.
+
+Use this package when switching on event schemas, comparing event kinds, or
+working with canonical WMO condition codes.
+
+## Event Kind Constants
+
+Current event kind constants are:
+
+| Constant | Value |
+|---|---|
+| `KindObservation` | `observation` |
+| `KindForecast` | `forecast` |
+| `KindForecastDiscussion` | `forecast_discussion` |
+| `KindWeatherStory` | `weather_story` |
+| `KindAlert` | `alert` |
+| `KindOutlook` | `outlook` |
+
+These are plain string constants. Convert them at adapter boundaries when using
+feedkit's `event.Kind` type.
+
+## Canonical Schema Constants
+
+Canonical schemas emitted after normalization:
+
+| Constant | Value |
+|---|---|
+| `SchemaWeatherObservationV1` | `weather.observation.v1` |
+| `SchemaWeatherForecastV1` | `weather.forecast.v1` |
+| `SchemaWeatherForecastDiscussionV1` | `weather.forecast_discussion.v1` |
+| `SchemaWeatherStoryV1` | `weather.weather_story.v1` |
+| `SchemaWeatherAlertV1` | `weather.alert.v1` |
+| `SchemaWeatherOutlookV1` | `weather.outlook.v1` |
+
+## Raw Schema Constants
+
+Raw source schemas emitted by current registered sources:
+
+| Constant | Value |
+|---|---|
+| `SchemaRawNWSObservationV1` | `raw.nws.observation.v1` |
+| `SchemaRawOpenMeteoCurrentV1` | `raw.openmeteo.current.v1` |
+| `SchemaRawOpenWeatherCurrentV1` | `raw.openweather.current.v1` |
+| `SchemaRawNWSHourlyForecastV1` | `raw.nws.hourly.forecast.v1` |
+| `SchemaRawNWSNarrativeForecastV1` | `raw.nws.narrative.forecast.v1` |
+| `SchemaRawNWSForecastDiscussionV1` | `raw.nws.forecast_discussion.v1` |
+| `SchemaRawNWSWeatherStoriesV1` | `raw.nws.weatherstories.v1` |
+| `SchemaRawOpenMeteoHourlyForecastV1` | `raw.openmeteo.hourly.forecast.v1` |
+| `SchemaRawNWSAlertsV1` | `raw.nws.alerts.v1` |
+| `SchemaRawSPCConvectiveOutlookV1` | `raw.spc.convective_outlook.v1` |
+
+Additional raw schema constant:
+
+| Constant | Value |
+|---|---|
+| `SchemaRawOpenWeatherHourlyForecastV1` | `raw.openweather.hourly.forecast.v1` |
+
+`SchemaRawOpenWeatherHourlyForecastV1` exists in code, but no current registered
+source emits it. Consumers should not expect that raw schema unless a later
+registered source documents it as part of the current event contract.
+
+## WMO Constants And Text
+
+`standards` also defines the canonical `WMOCode` vocabulary and text helpers
+used by normalized observations and forecasts.
+
+Consumer guidance:
+
+- Treat `WMOUnknown` as unknown condition data.
+- Observation `conditionCode` is required in the current event contract.
+- Forecast period `conditionCode` is optional because some forecast products do
+  not provide a meaningful WMO condition.
+- Prefer WMO constants and helper functions from this package instead of
+  duplicating code tables in consumers.
+
+## Boundaries
+
+`standards` is provider-agnostic. Provider-specific parsing belongs in
+`weatherfeeder` internals, not in this package and not in consumers.

docs/integrations/feedkit.md

@@ -0,0 +1,106 @@
+# Feedkit Integration
+
+## Purpose
+
+This document describes the feedkit runtime behavior that `weatherfeeder`
+currently depends on. It is for maintainers and LLM coding agents changing
+runtime wiring, config behavior, source construction, processing, routing, or
+sink behavior.
+
+Weather-domain behavior belongs in `weatherfeeder`. Generic daemon mechanics
+belong to feedkit.
+
+## Current Dependency
+
+`weatherfeeder` imports feedkit as its daemon framework dependency. The exact
+module version is declared in `go.mod`.
+
+Feedkit provides:
+
+- YAML config loading and validation.
+- Source, processor, and sink registries.
+- HTTP source helper behavior.
+- Scheduler polling.
+- Normalize and dedupe processors.
+- Route compilation and sink dispatch.
+- Built-in stdout, NATS, and Postgres sink mechanics.
+
+## Config Contract
+
+`cmd/weatherfeeder` calls feedkit config loading for `config.yml` in the current
+working directory.
+
+Implemented behavior relied on by weatherfeeder docs and tests:
+
+- Top-level config contains `sources`, `sinks`, and optional `routes`.
+- Config struct fields are decoded strictly, so misspelled struct fields fail
+  startup.
+- Driver-specific `params` maps are decoded generically and validated by the
+  source or sink constructor that consumes them.
+- Source `kinds` can be validated against a source's advertised `Kinds()`.
+
+## Source And HTTP Contract
+
+Most weatherfeeder sources use feedkit's single-document HTTP source helper for:
+
+- request construction;
+- `User-Agent` and `Accept` headers;
+- optional conditional GET validators;
+- response body size limits;
+- context-aware HTTP work;
+- unchanged `304 Not Modified` responses that emit no events.
+
+The SPC convective outlook source fetches multiple documents itself, but it uses
+feedkit transport helpers for HTTP clients and response body limits.
+
+## Scheduler And Processing Contract
+
+Weatherfeeder builds feedkit scheduler jobs from source configs. Current source
+drivers are polling drivers and use the configured `every` interval.
+
+Events flow through a feedkit pipeline in this order:
+
+1. normalize processor;
+2. dedupe processor.
+
+The normalize processor is configured with `RequireMatch=false`, so unmatched
+schemas pass through unchanged. Weatherfeeder registers its built-in normalizers
+and owns the provider-to-canonical mapping.
+
+The dedupe processor stores a bounded in-memory set of recent event IDs. The
+bound is configured in `cmd/weatherfeeder`.
+
+## Dispatch And Sink Contract
+
+Feedkit compiles routes from config and dispatches processed events to matching
+sinks. If `routes` is omitted, every configured sink receives every event kind.
+
+Feedkit owns sink fanout mechanics, per-sink workers, queueing, context-aware
+shutdown, and sink error logging. Weatherfeeder owns the event kinds and schemas
+that make routes meaningful.
+
+Built-in feedkit sinks used by weatherfeeder:
+
+- `stdout`: validates and writes JSON events to stdout.
+- `nats`: publishes JSON events to a configured subject.
+- generic `postgres` sink factory: opens the database, ensures tables/indexes,
+  runs transactions, inserts mapped rows, and prunes when configured.
+
+Weatherfeeder supplies its Postgres table schema and event mapper to feedkit's
+Postgres sink factory. The table contract is documented in
+[`postgres.md`](postgres.md).
+
+## Boundaries
+
+Do not move weather-domain policy into feedkit. Weatherfeeder owns:
+
+- provider source drivers;
+- raw and canonical schema constants;
+- event kind meaning;
+- canonical payload structs;
+- normalizers;
+- Postgres table shape and row mapping.
+
+Do not duplicate generic feedkit mechanics in weatherfeeder unless there is a
+narrow weather-specific reason. Runtime composition details are documented in
+[`../internal/runtime.md`](../internal/runtime.md).

docs/integrations/spc.md

@@ -39,11 +39,11 @@ current Day 1-3 SPC product URLs.
 
 ## Upstream Products Used
 
-The source fetches twelve required GeoJSON products every poll:
+The source fetches nine required GeoJSON products every poll:
 
 - Day 1 categorical, tornado, hail, and wind
 - Day 2 categorical, tornado, hail, and wind
-- Day 3 categorical, tornado, hail, and wind
+- Day 3 categorical
 
 It also fetches three required print pages:
 

docs/internal/runtime.md

@@ -68,6 +68,21 @@ Runtime composition uses feedkit for:
 
 Weatherfeeder registers its own source drivers and its Postgres schema mapper.
 
+Responsibility split:
+
+| Runtime concern | Owner |
+| --- | --- |
+| Config loading and generic validation | feedkit |
+| Source, processor, and sink registries | feedkit mechanics; weatherfeeder registrations |
+| Source polling and stream supervision | feedkit scheduler |
+| Raw weather data fetching | weatherfeeder source adapters |
+| Normalizer execution order and pass-through behavior | feedkit normalize processor |
+| Weather raw-to-canonical mapping | weatherfeeder normalizers |
+| Dedupe mechanics | feedkit dedupe processor |
+| Route compilation and sink fanout | feedkit dispatch |
+| Weather Postgres table shape and row mapping | weatherfeeder Postgres adapter |
+| Postgres connection, DDL, inserts, transactions, and pruning | feedkit Postgres sink |
+
 ## State
 
 Weatherfeeder-owned runtime state is in process:

docs/policy/architecture.md

@@ -23,6 +23,27 @@ The implemented runtime flow is:
 
 Canonical payload structs live in `model`. Schema identifiers and cross-provider wire conventions live in `standards`. Source adapters live under `internal/sources`. Normalizers live under `internal/normalizers`. Provider-specific parsing helpers shared by sources and normalizers live under `internal/providers`. Sink-specific persistence mapping lives under `internal/sinks`.
 
+## Architecture Style
+
+`weatherfeeder` uses a pragmatic ports-and-adapters architecture rather than a
+formal framework. Provider APIs, config loading, scheduling, dispatch, and sinks
+sit outside the weather domain model and normalization rules.
+
+The implementation style is:
+
+- Pipeline-oriented: events flow from source polling through normalization,
+  dedupe, routing, and sink fanout.
+- Schema-routed: normalizers select raw payloads by explicit schema strings, not
+  source names or configured routes.
+- Provider-isolated: NWS, Open-Meteo, OpenWeather, and SPC quirks stay in
+  provider-specific source, provider-helper, and normalizer packages.
+- Registry-based: built-in source drivers, normalizers, processors, and sinks
+  are assembled explicitly through registries instead of dynamic plugin loading.
+- Adapter-clean: persistence and external-system details stay behind source and
+  sink adapters, not in `model` or normalizers.
+- Direct Go: prefer small package-level constructors and straightforward code
+  over broad abstractions.
+
 ## Core Design Principles
 
 - Hexagonal boundaries: provider APIs, config loading, scheduling, dispatch, and sinks are external mechanisms around the weather domain model and normalization logic.
@@ -57,6 +78,23 @@ Tests and examples:
 - The sample `cmd/weatherfeeder/config.yml` is executable test input and is load-tested.
 - Tests should keep exercising package contracts directly rather than relying only on full-daemon execution.
 
+## Feedkit Boundary
+
+`feedkit` provides reusable daemon infrastructure. `weatherfeeder` provides the
+weather-domain adapters, models, schemas, and mapping policy.
+
+| Area | Feedkit owns | Weatherfeeder owns |
+| --- | --- | --- |
+| Config | Generic YAML shape: sources, sinks, routes, modes, cadence, and params. | Driver-specific config rules such as NWS `user_agent`, OpenWeather `units=metric`, and SPC coordinates. |
+| Events | Domain-agnostic event envelope: ID, kind, source, emitted/effective times, schema, and payload. | Event kind meaning, schema strings, and canonical weather payloads. |
+| Sources | Source interfaces, registry, expected-kind validation, HTTP helper, and default event ID helper. | NWS/Open-Meteo/OpenWeather/SPC source drivers and raw schema emission. |
+| Processing | Processor registry, normalize processor, dedupe processor, and pipeline execution. | Weather normalizers and schema-specific raw-to-canonical mapping. |
+| Dispatch | Route compilation and sink fanout mechanics. | Which weather event kinds are configured and meaningful. |
+| Sinks | Generic stdout, NATS, and Postgres sink mechanics. | Weather-specific Postgres schema and canonical event-to-row mapping. |
+
+Do not move weather-domain policy into `feedkit`, and do not duplicate generic
+daemon mechanics in `weatherfeeder` when feedkit already provides the boundary.
+
 ## Modules Or Processing Steps
 
 The implemented processing steps are source polling, normalization, dedupe, and sink dispatch.

docs/policy/development.md

@@ -24,7 +24,8 @@ docs, not here.
   normalizers.
 - `internal/sinks/postgres/`: weatherfeeder-owned Postgres schema and canonical
   event mapper.
-- `docs/`: current behavior, policies, integration contracts, and roadmap files.
+- `docs/`: current behavior, consumer guides, integration contracts, policies,
+  and roadmap files.
 - `examples/`: maintained, copyable configuration examples.
 
 ## Build And Test
@@ -70,6 +71,32 @@ Use fixtures, local test servers, and package-level tests.
 - Prefer explicit registries and small package-level constructors over hidden
   global behavior.
 
+## Architecture-Preserving Changes
+
+When changing `weatherfeeder`, preserve the split between feedkit
+infrastructure and weather-domain behavior.
+
+Do:
+
+- keep generic scheduling, dispatch, processor, config, and sink mechanics in
+  feedkit;
+- keep weather provider rules in source adapters, provider helpers, and
+  normalizers;
+- keep canonical weather payloads in `model` and schema/wire identifiers in
+  `standards`;
+- keep Postgres table and row mapping under `internal/sinks/postgres`;
+- use explicit registries for built-in sources and normalizers.
+
+Do not:
+
+- move provider parsing, WMO mapping, or canonical weather policy into
+  `cmd/weatherfeeder`;
+- move weather-specific constants, schemas, or validation rules into feedkit;
+- put database column metadata or sink-specific tags on canonical model structs;
+- replace explicit registries with dynamic plugin loading;
+- introduce broad abstractions when a small provider-specific helper preserves
+  clarity.
+
 ## Dependency Policy
 
 Prefer the Go standard library unless a dependency materially improves
@@ -183,6 +210,7 @@ When behavior changes, update the canonical docs in the same change:
 - CLI behavior: `docs/cli.md`;
 - operations and recovery: `docs/operations.md`;
 - troubleshooting: `docs/troubleshooting.md`;
+- public Go package consumption: `docs/consumers/`;
 - external contracts: `docs/integrations/`;
 - internal component behavior: `docs/internal/`;
 - copyable configs: `examples/`.

docs/policy/documentation.md

@@ -2,12 +2,13 @@
 
 ## Purpose
 
-Project documentation must help four audiences:
+Project documentation must help five audiences:
 
 1. users who need to run the application;
 2. administrators/operators who need to configure and operate it;
 3. developers who need to understand and change it safely;
-4. LLM coding agents that need clear scope, boundaries, and invariants.
+4. LLM coding agents that need clear scope, boundaries, and invariants;
+5. developers and LLM coding agents integrating this project from another codebase.
 
 Docs should be accurate, concise, task-oriented, and organized by audience. Prefer links to canonical docs over repetition.
 
@@ -42,11 +43,14 @@ Canonical homes:
 
 - project purpose and quickstart: `README.md`
 - development principles: `docs/policy/architecture.md`
+- public HTTP API reference: `docs/api.md`
 - configuration reference: `docs/config.md`
 - CLI reference: `docs/cli.md`
 - operations and recovery: `docs/operations.md`
 - troubleshooting: `docs/troubleshooting.md`
+- public API/package consumer guidance: `docs/consumers/`
 - implemented internals: `docs/internal/`
+- external protocol, service, and file-format contracts: `docs/integrations/`
 - future work: `docs/roadmap/`
 - contributor workflow: `docs/policy/development.md`
 - copyable examples: `examples/`
@@ -106,7 +110,7 @@ Recommended:
 - `examples/`
 - `docs/policy/development.md`
 
-### Modular, staged, service-oriented, or orchestration application
+### Modular, service-oriented, or orchestration application
 
 Required:
 - `docs/cli.md`, if CLI-based
@@ -119,6 +123,31 @@ Recommended:
 - `docs/troubleshooting.md`
 - validated examples under `examples/`
 
+### Public HTTP API service
+
+Required:
+- `docs/api.md`
+- `docs/cli.md`, if CLI-based
+- `docs/config.md`, if config-driven
+- `docs/operations.md`
+- `docs/internal/`
+- `docs/policy/development.md`
+
+Recommended:
+- `docs/troubleshooting.md`
+- `docs/consumers/`, for task-oriented client integration guides
+- `docs/integrations/`, for upstream/downstream service contracts
+- validated examples under `examples/`
+
+### Project with public packages or consumer APIs
+
+Required:
+- `docs/consumers/api.md`
+- one `docs/consumers/pkg-<name>.md` file per public package, if public packages exist
+
+Recommended:
+- copyable consumer examples under `examples/`, if practical
+
 ## Required Documents
 
 ### README.md
@@ -161,6 +190,32 @@ It should include:
 
 For small projects, this file may be brief. It may simply state that the project is intentionally narrow, monolithic, and dependency-light.
 
+### docs/api.md
+
+**Audience:** external HTTP API consumers, developers, LLM coding agents integrating by HTTP
+
+Required for projects whose primary public interface is HTTP.
+
+`docs/api.md` is the canonical public HTTP API contract. It should be normative for external consumers and should not be duplicated by README, operations docs, consumer guides, or integration docs.
+
+It should include:
+
+1. base URL conventions;
+2. authentication and authorization behavior, if implemented;
+3. response envelope;
+4. supported media types and content negotiation behavior;
+5. shared query parameters;
+6. endpoint reference grouped by route family;
+7. request parameters and validation rules;
+8. response fields, units, nullability, and optionality;
+9. error response shape and status codes;
+10. pagination, caching, rate-limit, idempotency, and retry behavior, if implemented;
+11. compact request and response examples.
+
+It must document only implemented endpoints and behavior. Planned endpoints, proposed fields, future filters, and experimental response shapes belong only under `docs/roadmap/`.
+
+For HTTP API projects, `docs/consumers/` may provide task-oriented client integration guides, but those guides should link to `docs/api.md` for the authoritative endpoint contract.
+
 ### docs/policy/development.md
 
 **Audience:** developers, LLM coding agents
@@ -175,7 +230,7 @@ It should include:
 - dependency policy;
 - how to add config fields;
 - how to add CLI flags;
-- how to add stages/modules/adapters, if applicable;
+- how to add modules or adapters, if applicable;
 - how to update examples;
 - documentation update expectations.
 
@@ -216,7 +271,7 @@ Explain when commands are useful, not just their syntax.
 
 **Audience:** administrators, operators
 
-Required for applications that maintain state, support resume behavior, run multiple stages, write durable artifacts, use remote storage, or require recovery procedures.
+Required for applications that maintain state, support resume behavior, run multi-step workflows, write durable artifacts, use remote storage, or require recovery procedures.
 
 It should cover:
 
@@ -244,11 +299,40 @@ Each entry should include:
 - safe fix;
 - relevant links.
 
+### docs/consumers/
+
+**Audience:** developers and LLM coding agents integrating this project from another codebase
+
+Required for projects with public packages, SDKs, client APIs, plugin APIs, or other application-facing integration surfaces.
+
+This directory describes how an external codebase should consume the project's public API. It should be task-oriented and copyable where useful. It is not the place for internal implementation details or operator procedures.
+
+For projects whose public API is HTTP, `docs/consumers/` is not required, and it should not duplicate the endpoint reference in `docs/api.md`. If present, it may provide practical integration workflows, client-specific examples, or migration notes that link back to `docs/api.md`.
+
+`docs/consumers/api.md` should provide the consumer-facing overview and primary implementation workflow. It should include:
+
+1. intended consumer audience and use cases;
+2. required inputs supplied by operators or deployment configuration;
+3. recommended public package or API workflow;
+4. minimal copyable example;
+5. consumer responsibilities and boundaries;
+6. retry, idempotency, or status behavior, if applicable;
+7. links to package-specific docs and canonical integration contracts.
+
+Package-specific docs should be named `pkg-<name>.md` and should include:
+
+1. import path;
+2. intended use cases;
+3. primary types and functions needed by consumers;
+4. minimal examples;
+5. validation, error, retry, and boundary behavior;
+6. links to canonical file-format or wire-protocol contracts.
+
 ### docs/internal/
 
 **Audience:** developers, LLM coding agents
 
-Required for modular, staged, service-oriented, or orchestration projects.
+Required for modular, service-oriented, or orchestration projects.
 
 This directory describes implemented internal components. It is not the roadmap.
 
@@ -289,7 +373,9 @@ Roadmap docs should not be confused with current behavior.
 
 Required for projects that depend on external CLIs, APIs, services, protocols, or file formats where the integration contract is important to maintain.
 
-This directory contains concise, versioned reference notes for external integration contracts. It should document only the parts of the external system that this project actually uses.
+This directory contains concise, versioned reference notes for external integration contracts. It should document only the parts of the external system that this project actually uses or exposes.
+
+For public HTTP API services, `docs/integrations/` should document upstream, downstream, storage, protocol, or runtime contracts that the service depends on or bridges. It should not become a second copy of the public HTTP endpoint reference; that belongs in `docs/api.md`.
 
 Use one file per integration where useful.
 
@@ -346,8 +432,10 @@ Before merging documentation changes, verify:
 
 - README is concise and orientation-focused.
 - `docs/policy/architecture.md` describes development principles.
+- `docs/api.md` is the canonical HTTP contract for HTTP API services.
 - Future work appears only under `docs/roadmap/`.
 - User-facing docs avoid unnecessary internals.
+- Consumer-facing docs explain public APIs without duplicating HTTP endpoint or integration contracts.
 - Developer-facing docs preserve boundaries and invariants.
 - Config examples match the schema.
 - CLI examples match real commands and flags.

internal/normalizers/common/finalize_test.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
 )
 
 func TestFinalizeRoundsWeatherPayloadFloats(t *testing.T) {
@@ -14,7 +15,7 @@ func TestFinalizeRoundsWeatherPayloadFloats(t *testing.T) {
 
 	in := event.Event{
 		ID:        "evt-1",
-		Kind:      event.Kind("observation"),
+		Kind:      event.Kind(standards.KindObservation),
 		Source:    "source-a",
 		EmittedAt: time.Date(2026, 3, 28, 12, 0, 0, 0, time.UTC),
 		Schema:    "raw.example.v1",

internal/normalizers/common/json.go

@@ -18,6 +18,18 @@ import (
 // Errors include a small amount of operation context ("extract payload", "decode raw payload").
 // Callers typically wrap these with a provider/kind label.
 func DecodeJSONPayload[T any](in event.Event) (T, error) {
+	var zero T
+
+	if typed, ok := in.Payload.(T); ok {
+		return typed, nil
+	}
+	if ptr, ok := in.Payload.(*T); ok {
+		if ptr == nil {
+			return zero, fmt.Errorf("extract payload: payload pointer is nil")
+		}
+		return *ptr, nil
+	}
+
 	return fknormalize.DecodeJSONPayload[T](in)
 }
 

internal/normalizers/common/json_test.go

@@ -0,0 +1,39 @@
+package common
+
+import (
+	"testing"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+func TestDecodeJSONPayloadAcceptsTypedPayload(t *testing.T) {
+	type rawPayload struct {
+		Value string `json:"value"`
+	}
+
+	got, err := DecodeJSONPayload[rawPayload](event.Event{
+		Payload: rawPayload{Value: "ok"},
+	})
+	if err != nil {
+		t.Fatalf("DecodeJSONPayload() error = %v", err)
+	}
+	if got.Value != "ok" {
+		t.Fatalf("Value = %q, want ok", got.Value)
+	}
+}
+
+func TestDecodeJSONPayloadAcceptsTypedPointerPayload(t *testing.T) {
+	type rawPayload struct {
+		Value string `json:"value"`
+	}
+
+	got, err := DecodeJSONPayload[rawPayload](event.Event{
+		Payload: &rawPayload{Value: "ok"},
+	})
+	if err != nil {
+		t.Fatalf("DecodeJSONPayload() error = %v", err)
+	}
+	if got.Value != "ok" {
+		t.Fatalf("Value = %q, want ok", got.Value)
+	}
+}

internal/normalizers/nws/forecast_discussion_test.go

@@ -18,7 +18,7 @@ func TestForecastDiscussionNormalizerProducesCanonicalSchema(t *testing.T) {
 
 	out, err := (ForecastDiscussionNormalizer{}).Normalize(nil, event.Event{
 		ID:        "evt-discussion-1",
-		Kind:      event.Kind("forecast_discussion"),
+		Kind:      event.Kind(standards.KindForecastDiscussion),
 		Source:    "nws-discussion-test",
 		EmittedAt: time.Date(2026, 3, 28, 19, 25, 0, 0, time.UTC),
 		Schema:    standards.SchemaRawNWSForecastDiscussionV1,
@@ -33,7 +33,7 @@ func TestForecastDiscussionNormalizerProducesCanonicalSchema(t *testing.T) {
 	if out.Schema != standards.SchemaWeatherForecastDiscussionV1 {
 		t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherForecastDiscussionV1)
 	}
-	if out.Kind != event.Kind("forecast_discussion") {
+	if out.Kind != event.Kind(standards.KindForecastDiscussion) {
 		t.Fatalf("Kind = %q, want forecast_discussion", out.Kind)
 	}
 
@@ -74,7 +74,7 @@ func TestForecastDiscussionNormalizerProducesCanonicalSchema(t *testing.T) {
 func TestForecastDiscussionNormalizerRejectsMissingIssueTime(t *testing.T) {
 	_, err := (ForecastDiscussionNormalizer{}).Normalize(nil, event.Event{
 		ID:        "evt-discussion-bad",
-		Kind:      event.Kind("forecast_discussion"),
+		Kind:      event.Kind(standards.KindForecastDiscussion),
 		Source:    "nws-discussion-test",
 		EmittedAt: time.Date(2026, 3, 28, 19, 25, 0, 0, time.UTC),
 		Schema:    standards.SchemaRawNWSForecastDiscussionV1,
@@ -93,7 +93,7 @@ func TestForecastDiscussionNormalizerWireShapeHasNoUnexpectedKeys(t *testing.T)
 
 	out, err := (ForecastDiscussionNormalizer{}).Normalize(nil, event.Event{
 		ID:        "evt-discussion-2",
-		Kind:      event.Kind("forecast_discussion"),
+		Kind:      event.Kind(standards.KindForecastDiscussion),
 		Source:    "nws-discussion-test",
 		EmittedAt: time.Date(2026, 3, 28, 19, 25, 0, 0, time.UTC),
 		Schema:    standards.SchemaRawNWSForecastDiscussionV1,

internal/normalizers/nws/forecast_test.go

@@ -182,7 +182,7 @@ func TestNormalizeForecastEventBySchemaProducesCanonicalWeatherForecastSchema(t
 		t.Run(tt.name, func(t *testing.T) {
 			out, err := normalizeForecastEventBySchema(event.Event{
 				ID:        "evt-1",
-				Kind:      event.Kind("forecast"),
+				Kind:      event.Kind(standards.KindForecast),
 				Source:    "nws-test",
 				EmittedAt: time.Date(2026, 3, 16, 18, 0, 0, 0, time.UTC),
 				Schema:    tt.schema,

internal/normalizers/nws/weatherstories_test.go

@@ -22,7 +22,7 @@ func TestWeatherStoriesNormalizerProducesCanonicalSchemaAndMapsSample(t *testing
 	if out.Schema != standards.SchemaWeatherStoryV1 {
 		t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherStoryV1)
 	}
-	if out.Kind != event.Kind("weather_story") {
+	if out.Kind != event.Kind(standards.KindWeatherStory) {
 		t.Fatalf("Kind = %q, want weather_story", out.Kind)
 	}
 
@@ -118,7 +118,7 @@ func TestWeatherStoriesNormalizerMatch(t *testing.T) {
 func weatherStoriesRawEvent(payload string) event.Event {
 	return event.Event{
 		ID:        "evt-weatherstories-1",
-		Kind:      event.Kind("weather_story"),
+		Kind:      event.Kind(standards.KindWeatherStory),
 		Source:    "nws-weatherstories-test",
 		EmittedAt: time.Date(2026, 5, 30, 9, 5, 0, 0, time.UTC),
 		Schema:    standards.SchemaRawNWSWeatherStoriesV1,

internal/normalizers/spc/convective_outlook.go

@@ -22,7 +22,6 @@ const (
 	providerSPC        = "spc"
 	productConvective  = "convective"
 	outlookNormalizer  = "spc convective outlook"
-	outlookKind        = "outlook"
 	outlookTypeUnknown = 99
 )
 

internal/normalizers/spc/convective_outlook_test.go

@@ -2,8 +2,6 @@ package spc
 
 import (
 	"encoding/json"
-	"os"
-	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -32,7 +30,7 @@ func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *test
 	if out.Schema != standards.SchemaWeatherOutlookV1 {
 		t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherOutlookV1)
 	}
-	if out.Kind != event.Kind("outlook") {
+	if out.Kind != event.Kind(standards.KindOutlook) {
 		t.Fatalf("Kind = %q, want outlook", out.Kind)
 	}
 
@@ -56,8 +54,8 @@ func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *test
 	if run.Latitude == nil || *run.Latitude != 38.5 || run.Longitude == nil || *run.Longitude != -90.5 {
 		t.Fatalf("coordinates = %v,%v", run.Latitude, run.Longitude)
 	}
-	if len(run.Outlooks) != 12 {
+	if len(run.Outlooks) != 9 {
-		t.Fatalf("Outlooks length = %d, want 12", len(run.Outlooks))
+		t.Fatalf("Outlooks length = %d, want 9", len(run.Outlooks))
 	}
 
 	got := run.Outlooks[0]
@@ -107,6 +105,21 @@ func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *test
 	}
 }
 
+func TestConvectiveOutlookNormalizerAcceptsTypedSourcePayload(t *testing.T) {
+	bundle := spcBundle(t, 38.5, -90.5)
+	in := spcRawEvent(t, bundle)
+	in.Payload = bundle
+
+	out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, in)
+	if err != nil {
+		t.Fatalf("Normalize() error = %v", err)
+	}
+	run := out.Payload.(model.WeatherOutlookRun)
+	if len(run.Outlooks) != 9 {
+		t.Fatalf("Outlooks length = %d, want 9", len(run.Outlooks))
+	}
+}
+
 func TestConvectiveOutlookNormalizerOrdersProductsByDayAndType(t *testing.T) {
 	bundle := spcBundle(t, 0, 0)
 	for i, j := 0, len(bundle.Products)-1; i < j; i, j = i+1, j-1 {
@@ -276,7 +289,7 @@ func spcRawEvent(t *testing.T, bundle spcprovider.RawConvectiveOutlookBundle) ev
 	effectiveAt := time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC)
 	return event.Event{
 		ID:          "evt-spc-outlook-1",
-		Kind:        event.Kind("outlook"),
+		Kind:        event.Kind(standards.KindOutlook),
 		Source:      "spc-test",
 		EmittedAt:   time.Date(2026, 6, 11, 20, 5, 0, 0, time.UTC),
 		EffectiveAt: &effectiveAt,
@@ -322,23 +335,13 @@ func geoJSONFixtureForProduct(t *testing.T, key string) []byte {
 	case strings.HasPrefix(key, "day2_"):
 		return readSPCTestFixture(t, "day2_torn.geojson")
 	case strings.HasPrefix(key, "day3_"):
-		return readSPCTestFixture(t, "day3_wind.geojson")
+		return readSPCTestFixture(t, "day3_cat.geojson")
 	default:
 		t.Fatalf("unknown product key %q", key)
 		return nil
 	}
 }
 
-func readSPCTestFixture(t *testing.T, name string) []byte {
-	t.Helper()
-	path := filepath.Join("..", "..", "providers", "spc", "testdata", name)
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		t.Fatalf("read fixture %s: %v", path, err)
-	}
-	return raw
-}
-
 func findOutlook(outlooks []model.WeatherOutlook, day int, outlookType string) *model.WeatherOutlook {
 	for i := range outlooks {
 		if outlooks[i].Day == day && outlooks[i].OutlookType == outlookType {

internal/normalizers/spc/fixture_test.go

@@ -0,0 +1,17 @@
+package spc
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func readSPCTestFixture(t *testing.T, name string) []byte {
+	t.Helper()
+	path := filepath.Join("..", "..", "providers", "spc", "testdata", name)
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read fixture %s: %v", path, err)
+	}
+	return raw
+}

internal/providers/spc/fixture_test.go

@@ -0,0 +1,17 @@
+package spc
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func readTestFile(t *testing.T, name string) []byte {
+	t.Helper()
+	path := filepath.Join("testdata", name)
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read %s: %v", path, err)
+	}
+	return raw
+}

internal/providers/spc/geojson_test.go

@@ -1,8 +1,6 @@
 package spc
 
 import (
-	"os"
-	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -78,13 +76,3 @@ func TestParseISOTimestampTrimsAndReturnsUTC(t *testing.T) {
 		t.Fatalf("ParseISOTimestamp() = %s, want %s", got, want)
 	}
 }
-
-func readTestFile(t *testing.T, name string) []byte {
-	t.Helper()
-	path := filepath.Join("testdata", name)
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		t.Fatalf("read %s: %v", path, err)
-	}
-	return raw
-}

internal/providers/spc/product.go

@@ -34,9 +34,6 @@ var geoJSONProducts = []GeoJSONProduct{
 	{Key: "day2_hail", Day: 2, OutlookType: OutlookTypeHail, URL: "https://www.spc.noaa.gov/products/outlook/day2otlk_hail.nolyr.geojson"},
 	{Key: "day2_wind", Day: 2, OutlookType: OutlookTypeWind, URL: "https://www.spc.noaa.gov/products/outlook/day2otlk_wind.nolyr.geojson"},
 	{Key: "day3_categorical", Day: 3, OutlookType: OutlookTypeCategorical, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_cat.nolyr.geojson"},
-	{Key: "day3_tornado", Day: 3, OutlookType: OutlookTypeTornado, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_torn.nolyr.geojson"},
-	{Key: "day3_hail", Day: 3, OutlookType: OutlookTypeHail, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_hail.nolyr.geojson"},
-	{Key: "day3_wind", Day: 3, OutlookType: OutlookTypeWind, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_wind.nolyr.geojson"},
 }
 
 var discussionProducts = []DiscussionProduct{

internal/providers/spc/product_test.go

@@ -4,8 +4,8 @@ import "testing"
 
 func TestGeoJSONProductsStableOrder(t *testing.T) {
 	got := GeoJSONProducts()
-	if len(got) != 12 {
+	if len(got) != 9 {
-		t.Fatalf("GeoJSONProducts() length = %d, want 12", len(got))
+		t.Fatalf("GeoJSONProducts() length = %d, want 9", len(got))
 	}
 
 	wantKeys := []string{
@@ -18,9 +18,6 @@ func TestGeoJSONProductsStableOrder(t *testing.T) {
 		"day2_hail",
 		"day2_wind",
 		"day3_categorical",
-		"day3_tornado",
-		"day3_hail",
-		"day3_wind",
 	}
 	for i, want := range wantKeys {
 		if got[i].Key != want {

internal/providers/spc/testdata/day3_cat.geojson

@@ -8,9 +8,9 @@
         "EXPIRE_ISO": "2026-06-14T12:00:00Z",
         "ISSUE_ISO": "2026-06-11T19:45:00Z",
         "FORECASTER": "LEE",
-        "LABEL": "15",
+        "LABEL": "MRGL",
-        "LABEL2": "15% Wind",
+        "LABEL2": "Marginal Risk",
-        "DN": 15
+        "DN": 2
       },
       "geometry": {
         "type": "MultiPolygon",

internal/sinks/postgres/map.go

@@ -48,13 +48,7 @@ func mapObservationEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 
 	writes = append(writes, fksinks.PostgresWrite{
 		Table: tableObservations,
-		Values: map[string]any{
+		Values: parentEventValues(e, map[string]any{
-			"event_id":                  e.ID,
-			"event_kind":                string(e.Kind),
-			"event_source":              e.Source,
-			"event_schema":              e.Schema,
-			"event_emitted_at":          e.EmittedAt.UTC(),
-			"event_effective_at":        nullableTime(e.EffectiveAt),
 			"station_id":                nullableString(obs.StationID),
 			"station_name":              nullableString(obs.StationName),
 			"observed_at":               observedAt,
@@ -70,7 +64,7 @@ func mapObservationEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 			"visibility_meters":         nullableFloat64(obs.VisibilityMeters),
 			"relative_humidity_percent": nullableFloat64(obs.RelativeHumidityPercent),
 			"apparent_temperature_c":    nullableFloat64(obs.ApparentTemperatureC),
-		},
+		}),
 	})
 
 	for i, pw := range obs.PresentWeather {
@@ -109,13 +103,7 @@ func mapForecastEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 
 	writes = append(writes, fksinks.PostgresWrite{
 		Table: tableForecasts,
-		Values: map[string]any{
+		Values: parentEventValues(e, map[string]any{
-			"event_id":           e.ID,
-			"event_kind":         string(e.Kind),
-			"event_source":       e.Source,
-			"event_schema":       e.Schema,
-			"event_emitted_at":   e.EmittedAt.UTC(),
-			"event_effective_at": nullableTime(e.EffectiveAt),
 			"location_id":      nullableString(run.LocationID),
 			"location_name":    nullableString(run.LocationName),
 			"issued_at":        issuedAt,
@@ -125,7 +113,7 @@ func mapForecastEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 			"longitude":        nullableFloat64(run.Longitude),
 			"elevation_meters": nullableFloat64(run.ElevationMeters),
 			"period_count":     len(run.Periods),
-		},
+		}),
 	})
 
 	for i, p := range run.Periods {
@@ -186,13 +174,7 @@ func mapForecastDiscussionEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error
 	writes := make([]fksinks.PostgresWrite, 0, 1+len(run.KeyMessages))
 	writes = append(writes, fksinks.PostgresWrite{
 		Table: tableForecastDiscussions,
-		Values: map[string]any{
+		Values: parentEventValues(e, map[string]any{
-			"event_id":             e.ID,
-			"event_kind":           string(e.Kind),
-			"event_source":         e.Source,
-			"event_schema":         e.Schema,
-			"event_emitted_at":     e.EmittedAt.UTC(),
-			"event_effective_at":   nullableTime(e.EffectiveAt),
 			"office_id":            nullableString(run.OfficeID),
 			"office_name":          nullableString(run.OfficeName),
 			"issued_at":            issuedAt,
@@ -205,7 +187,7 @@ func mapForecastDiscussionEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error
 			"long_term_issued_at":  longTermIssuedAt,
 			"long_term_text":       longTermText,
 			"key_message_count":    len(run.KeyMessages),
-		},
+		}),
 	})
 
 	for i, msg := range run.KeyMessages {
@@ -236,17 +218,11 @@ func mapWeatherStoryEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 	writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Stories))
 	writes = append(writes, fksinks.PostgresWrite{
 		Table: tableWeatherStoryRuns,
-		Values: map[string]any{
+		Values: parentEventValues(e, map[string]any{
-			"event_id":           e.ID,
-			"event_kind":         string(e.Kind),
-			"event_source":       e.Source,
-			"event_schema":       e.Schema,
-			"event_emitted_at":   e.EmittedAt.UTC(),
-			"event_effective_at": nullableTime(e.EffectiveAt),
 			"office_id":   nullableString(run.OfficeID),
 			"as_of":       asOf,
 			"story_count": len(run.Stories),
-		},
+		}),
 	})
 
 	for i, story := range run.Stories {
@@ -290,20 +266,14 @@ func mapAlertEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 
 	writes = append(writes, fksinks.PostgresWrite{
 		Table: tableAlertRuns,
-		Values: map[string]any{
+		Values: parentEventValues(e, map[string]any{
-			"event_id":           e.ID,
-			"event_kind":         string(e.Kind),
-			"event_source":       e.Source,
-			"event_schema":       e.Schema,
-			"event_emitted_at":   e.EmittedAt.UTC(),
-			"event_effective_at": nullableTime(e.EffectiveAt),
 			"location_id":   nullableString(run.LocationID),
 			"location_name": nullableString(run.LocationName),
 			"as_of":         asOf,
 			"latitude":      nullableFloat64(run.Latitude),
 			"longitude":     nullableFloat64(run.Longitude),
 			"alert_count":   len(run.Alerts),
-		},
+		}),
 	})
 
 	for i, a := range run.Alerts {
@@ -372,13 +342,7 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 	writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Outlooks))
 	writes = append(writes, fksinks.PostgresWrite{
 		Table: tableOutlookRuns,
-		Values: map[string]any{
+		Values: parentEventValues(e, map[string]any{
-			"event_id":           e.ID,
-			"event_kind":         string(e.Kind),
-			"event_source":       e.Source,
-			"event_schema":       e.Schema,
-			"event_emitted_at":   e.EmittedAt.UTC(),
-			"event_effective_at": nullableTime(e.EffectiveAt),
 			"location_id":   nullableString(run.LocationID),
 			"location_name": nullableString(run.LocationName),
 			"latitude":      nullableFloat64(run.Latitude),
@@ -386,7 +350,7 @@ func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
 			"as_of":         asOf,
 			"issued_at":     nullableTime(run.IssuedAt),
 			"outlook_count": len(run.Outlooks),
-		},
+		}),
 	})
 
 	for i, outlook := range run.Outlooks {
@@ -488,6 +452,21 @@ func decodePayload[T any](payload any) (T, error) {
 	return out, nil
 }
 
+func parentEventValues(e fkevent.Event, values map[string]any) map[string]any {
+	out := map[string]any{
+		"event_id":           e.ID,
+		"event_kind":         string(e.Kind),
+		"event_source":       e.Source,
+		"event_schema":       e.Schema,
+		"event_emitted_at":   e.EmittedAt.UTC(),
+		"event_effective_at": nullableTime(e.EffectiveAt),
+	}
+	for k, v := range values {
+		out[k] = v
+	}
+	return out
+}
+
 func nullableDiscussionSection(section *model.WeatherForecastDiscussionSection) (any, any, any) {
 	if section == nil {
 		return nil, nil, nil

internal/sinks/postgres/map_test.go

@@ -27,7 +27,7 @@ func TestMapPostgresEventObservationStructPayload(t *testing.T) {
 		PresentWeather:  []model.PresentWeather{{Raw: map[string]any{"a": 1, "b": "x"}}},
 	}
 
-	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherObservationV1, "observation", obs))
+	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherObservationV1, standards.KindObservation, obs))
 	if err != nil {
 		t.Fatalf("mapPostgresEvent() error = %v", err)
 	}
@@ -74,7 +74,7 @@ func TestMapPostgresEventForecastStructPayload(t *testing.T) {
 		},
 	}
 
-	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, "forecast", run))
+	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, standards.KindForecast, run))
 	if err != nil {
 		t.Fatalf("mapPostgresEvent() error = %v", err)
 	}
@@ -122,7 +122,7 @@ func TestMapPostgresEventAlertStructPayload(t *testing.T) {
 		},
 	}
 
-	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherAlertV1, "alert", run))
+	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherAlertV1, standards.KindAlert, run))
 	if err != nil {
 		t.Fatalf("mapPostgresEvent() error = %v", err)
 	}
@@ -163,7 +163,7 @@ func TestMapPostgresEventForecastDiscussionStructPayload(t *testing.T) {
 		LongTerm:    &model.WeatherForecastDiscussionSection{Text: "Long term text"},
 	}
 
-	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastDiscussionV1, "forecast_discussion", run))
+	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastDiscussionV1, standards.KindForecastDiscussion, run))
 	if err != nil {
 		t.Fatalf("mapPostgresEvent() error = %v", err)
 	}
@@ -213,7 +213,7 @@ func TestMapPostgresEventWeatherStoryStructPayload(t *testing.T) {
 		},
 	}
 
-	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, "weather_story", run))
+	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, standards.KindWeatherStory, run))
 	if err != nil {
 		t.Fatalf("mapPostgresEvent() error = %v", err)
 	}
@@ -290,7 +290,7 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
 		},
 	}
 
-	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run))
+	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run))
 	if err != nil {
 		t.Fatalf("mapPostgresEvent() error = %v", err)
 	}
@@ -332,7 +332,7 @@ func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
 }
 
 func TestMapPostgresEventOutlookRejectsMissingAsOf(t *testing.T) {
-	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", model.WeatherOutlookRun{}))
+	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, model.WeatherOutlookRun{}))
 	if err == nil {
 		t.Fatalf("mapPostgresEvent() error = nil, want missing asOf error")
 	}
@@ -381,7 +381,7 @@ func TestMapPostgresEventOutlookRejectsMissingIDAndProvider(t *testing.T) {
 				AsOf:     time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
 				Outlooks: []model.WeatherOutlook{outlook},
 			}
-			_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run))
+			_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run))
 			if err == nil {
 				t.Fatalf("mapPostgresEvent() error = nil, want %q", tt.wantErr)
 			}
@@ -405,7 +405,7 @@ func TestMapPostgresEventOutlookRejectsMissingRequiredTimes(t *testing.T) {
 			Geometry:    json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`),
 		}},
 	}
-	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run))
+	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run))
 	if err == nil {
 		t.Fatalf("mapPostgresEvent() error = nil, want missing time error")
 	}
@@ -430,7 +430,7 @@ func TestMapPostgresEventOutlookRejectsEmptyGeometry(t *testing.T) {
 			ExpiresAt:   time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
 		}},
 	}
-	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run))
+	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, standards.KindOutlook, run))
 	if err == nil {
 		t.Fatalf("mapPostgresEvent() error = nil, want geometry error")
 	}
@@ -440,7 +440,7 @@ func TestMapPostgresEventOutlookRejectsEmptyGeometry(t *testing.T) {
 }
 
 func TestMapPostgresEventWeatherStoryRejectsMissingAsOf(t *testing.T) {
-	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, "weather_story", model.WeatherStoryRun{}))
+	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, standards.KindWeatherStory, model.WeatherStoryRun{}))
 	if err == nil {
 		t.Fatalf("mapPostgresEvent() error = nil, want missing asOf error")
 	}
@@ -454,7 +454,7 @@ func TestMapPostgresEventWeatherStoryRejectsMissingStoryTimes(t *testing.T) {
 		AsOf:    time.Date(2026, 5, 30, 9, 0, 34, 0, time.UTC),
 		Stories: []model.WeatherStory{{Title: "missing times"}},
 	}
-	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, "weather_story", run))
+	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, standards.KindWeatherStory, run))
 	if err == nil {
 		t.Fatalf("mapPostgresEvent() error = nil, want missing story times error")
 	}
@@ -484,7 +484,7 @@ func TestMapPostgresEventMapPayload(t *testing.T) {
 		t.Fatalf("json.Unmarshal() error = %v", err)
 	}
 
-	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, "forecast", payload))
+	writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, standards.KindForecast, payload))
 	if err != nil {
 		t.Fatalf("mapPostgresEvent() error = %v", err)
 	}
@@ -499,7 +499,7 @@ func TestMapPostgresEventMapPayload(t *testing.T) {
 }
 
 func TestMapPostgresEventUnknownSchemaNoOp(t *testing.T) {
-	writes, err := mapPostgresEvent(context.Background(), testEvent("weather.unknown.v1", "observation", map[string]any{"x": 1}))
+	writes, err := mapPostgresEvent(context.Background(), testEvent("weather.unknown.v1", standards.KindObservation, map[string]any{"x": 1}))
 	if err != nil {
 		t.Fatalf("mapPostgresEvent() error = %v", err)
 	}
@@ -509,7 +509,7 @@ func TestMapPostgresEventUnknownSchemaNoOp(t *testing.T) {
 }
 
 func TestMapPostgresEventMalformedPayload(t *testing.T) {
-	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, "forecast", "bad"))
+	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastV1, standards.KindForecast, "bad"))
 	if err == nil {
 		t.Fatalf("mapPostgresEvent() expected error for malformed payload")
 	}
@@ -519,7 +519,7 @@ func TestMapPostgresEventMalformedPayload(t *testing.T) {
 }
 
 func TestMapPostgresEventForecastDiscussionMalformedPayload(t *testing.T) {
-	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastDiscussionV1, "forecast_discussion", "bad"))
+	_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherForecastDiscussionV1, standards.KindForecastDiscussion, "bad"))
 	if err == nil {
 		t.Fatalf("mapPostgresEvent() expected error for malformed payload")
 	}
@@ -528,6 +528,56 @@ func TestMapPostgresEventForecastDiscussionMalformedPayload(t *testing.T) {
 	}
 }
 
+func TestParentEventValuesAddsEnvelopeAndPreservesProductValues(t *testing.T) {
+	emittedAt := time.Date(2026, 3, 16, 13, 31, 0, 0, time.FixedZone("CDT", -5*60*60))
+	effectiveAt := time.Date(2026, 3, 16, 13, 30, 0, 0, time.FixedZone("CDT", -5*60*60))
+	event := fkevent.Event{
+		ID:          "evt-envelope",
+		Kind:        fkevent.Kind(standards.KindForecast),
+		Source:      "test-source",
+		Schema:      standards.SchemaWeatherForecastV1,
+		EmittedAt:   emittedAt,
+		EffectiveAt: &effectiveAt,
+	}
+
+	got := parentEventValues(event, map[string]any{"product_col": "product-value"})
+
+	assertParentEnvelopeValues(t, got, event)
+	if got["product_col"] != "product-value" {
+		t.Fatalf("product_col = %#v, want product-value", got["product_col"])
+	}
+}
+
+func TestParentEventValuesNullEffectiveAt(t *testing.T) {
+	base := fkevent.Event{
+		ID:        "evt-envelope",
+		Kind:      fkevent.Kind(standards.KindObservation),
+		Source:    "test-source",
+		Schema:    standards.SchemaWeatherObservationV1,
+		EmittedAt: time.Date(2026, 3, 16, 18, 31, 0, 0, time.UTC),
+	}
+
+	for _, tt := range []struct {
+		name string
+		mut  func(*fkevent.Event)
+	}{
+		{name: "nil", mut: func(*fkevent.Event) {}},
+		{name: "zero", mut: func(event *fkevent.Event) {
+			zero := time.Time{}
+			event.EffectiveAt = &zero
+		}},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			event := base
+			tt.mut(&event)
+			got := parentEventValues(event, nil)
+			if got["event_effective_at"] != nil {
+				t.Fatalf("event_effective_at = %#v, want nil", got["event_effective_at"])
+			}
+		})
+	}
+}
+
 func testEvent(schema string, kind fkevent.Kind, payload any) fkevent.Event {
 	effectiveAt := time.Date(2026, 3, 16, 18, 30, 0, 0, time.UTC)
 	return fkevent.Event{
@@ -541,6 +591,30 @@ func testEvent(schema string, kind fkevent.Kind, payload any) fkevent.Event {
 	}
 }
 
+func assertParentEnvelopeValues(t *testing.T, values map[string]any, event fkevent.Event) {
+	t.Helper()
+
+	if got := values["event_id"]; got != event.ID {
+		t.Fatalf("event_id = %#v, want %q", got, event.ID)
+	}
+	if got := values["event_kind"]; got != string(event.Kind) {
+		t.Fatalf("event_kind = %#v, want %q", got, event.Kind)
+	}
+	if got := values["event_source"]; got != event.Source {
+		t.Fatalf("event_source = %#v, want %q", got, event.Source)
+	}
+	if got := values["event_schema"]; got != event.Schema {
+		t.Fatalf("event_schema = %#v, want %q", got, event.Schema)
+	}
+	if got := values["event_emitted_at"]; got != event.EmittedAt.UTC() {
+		t.Fatalf("event_emitted_at = %#v, want %s", got, event.EmittedAt.UTC())
+	}
+	wantEffective := nullableTime(event.EffectiveAt)
+	if got := values["event_effective_at"]; got != wantEffective {
+		t.Fatalf("event_effective_at = %#v, want %#v", got, wantEffective)
+	}
+}
+
 func firstWriteForTable(writes []fksinks.PostgresWrite, table string) (fksinks.PostgresWrite, bool) {
 	for _, w := range writes {
 		if w.Table == table {

internal/sinks/postgres/schema.go

@@ -26,13 +26,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 		Tables: []fksinks.PostgresTable{
 			{
 				Name: tableObservations,
-				Columns: []fksinks.PostgresColumn{
+				Columns: parentEnvelopeColumns([]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},
@@ -48,7 +42,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 					{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},
-				},
+				}...),
 				PrimaryKey:  []string{"event_id"},
 				PruneColumn: "observed_at",
 				Indexes: []fksinks.PostgresIndex{
@@ -73,13 +67,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 			},
 			{
 				Name: tableForecasts,
-				Columns: []fksinks.PostgresColumn{
+				Columns: parentEnvelopeColumns([]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},
@@ -89,7 +77,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 					{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{
@@ -137,13 +125,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 			},
 			{
 				Name: tableForecastDiscussions,
-				Columns: []fksinks.PostgresColumn{
+				Columns: parentEnvelopeColumns([]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: "office_id", Type: "TEXT", Nullable: true},
 					{Name: "office_name", Type: "TEXT", Nullable: true},
 					{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: false},
@@ -156,7 +138,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 					{Name: "long_term_issued_at", Type: "TIMESTAMPTZ", Nullable: true},
 					{Name: "long_term_text", Type: "TEXT", Nullable: true},
 					{Name: "key_message_count", Type: "INTEGER", Nullable: false},
-				},
+				}...),
 				PrimaryKey:  []string{"event_id"},
 				PruneColumn: "issued_at",
 				Indexes: []fksinks.PostgresIndex{
@@ -180,17 +162,11 @@ func PostgresSchema() fksinks.PostgresSchema {
 			},
 			{
 				Name: tableWeatherStoryRuns,
-				Columns: []fksinks.PostgresColumn{
+				Columns: parentEnvelopeColumns([]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: "office_id", Type: "TEXT", Nullable: true},
 					{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
 					{Name: "story_count", Type: "INTEGER", Nullable: false},
-				},
+				}...),
 				PrimaryKey:  []string{"event_id"},
 				PruneColumn: "as_of",
 				Indexes: []fksinks.PostgresIndex{
@@ -225,20 +201,14 @@ func PostgresSchema() fksinks.PostgresSchema {
 			},
 			{
 				Name: tableAlertRuns,
-				Columns: []fksinks.PostgresColumn{
+				Columns: parentEnvelopeColumns([]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{
@@ -301,13 +271,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 			},
 			{
 				Name: tableOutlookRuns,
-				Columns: []fksinks.PostgresColumn{
+				Columns: parentEnvelopeColumns([]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: "latitude", Type: "DOUBLE PRECISION", Nullable: true},
@@ -315,7 +279,7 @@ func PostgresSchema() fksinks.PostgresSchema {
 					{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
 					{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: true},
 					{Name: "outlook_count", Type: "INTEGER", Nullable: false},
-				},
+				}...),
 				PrimaryKey:  []string{"event_id"},
 				PruneColumn: "as_of",
 				Indexes: []fksinks.PostgresIndex{
@@ -362,3 +326,15 @@ func PostgresSchema() fksinks.PostgresSchema {
 		MapEvent: mapPostgresEvent,
 	}
 }
+
+func parentEnvelopeColumns(extra ...fksinks.PostgresColumn) []fksinks.PostgresColumn {
+	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},
+	}
+	return append(columns, extra...)
+}

internal/sinks/postgres/schema_test.go

@@ -1,8 +1,11 @@
 package postgres
 
 import (
+	"reflect"
 	"strings"
 	"testing"
+
+	fksinks "gitea.maximumdirect.net/ejr/feedkit/sinks"
 )
 
 func TestWeatherPostgresSchemaShape(t *testing.T) {
@@ -88,6 +91,28 @@ func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) {
 	}
 }
 
+func TestWeatherPostgresSchemaParentTablesStartWithEnvelopeColumns(t *testing.T) {
+	for _, table := range []string{
+		tableObservations,
+		tableForecasts,
+		tableForecastDiscussions,
+		tableWeatherStoryRuns,
+		tableAlertRuns,
+		tableOutlookRuns,
+	} {
+		t.Run(table, func(t *testing.T) {
+			columns := orderedColumnsForTable(t, table)
+			want := parentEnvelopeColumns()
+			if len(columns) < len(want) {
+				t.Fatalf("%s has %d columns, want at least %d", table, len(columns), len(want))
+			}
+			if !reflect.DeepEqual(columns[:len(want)], want) {
+				t.Fatalf("%s envelope prefix = %#v, want %#v", table, columns[:len(want)], want)
+			}
+		})
+	}
+}
+
 func assertTablePrimaryKey(t *testing.T, table string, want []string) {
 	t.Helper()
 	for _, tbl := range PostgresSchema().Tables {
@@ -121,20 +146,26 @@ func assertTableIndex(t *testing.T, table string, name string, want []string) {
 	t.Fatalf("missing table %q", table)
 }
 
-func columnsForTable(t *testing.T, table string) map[string]bool {
+func orderedColumnsForTable(t *testing.T, table string) []fksinks.PostgresColumn {
 	t.Helper()
 
 	schema := PostgresSchema()
 	for _, tbl := range schema.Tables {
-		if tbl.Name != table {
+		if tbl.Name == table {
-			continue
+			return tbl.Columns
 		}
-		cols := make(map[string]bool, len(tbl.Columns))
-		for _, col := range tbl.Columns {
-			cols[col.Name] = true
-		}
-		return cols
 	}
 	t.Fatalf("missing table %q", table)
 	return nil
 }
+
+func columnsForTable(t *testing.T, table string) map[string]bool {
+	t.Helper()
+
+	ordered := orderedColumnsForTable(t, table)
+	cols := make(map[string]bool, len(ordered))
+	for _, col := range ordered {
+		cols[col.Name] = true
+	}
+	return cols
+}

internal/sources/builtins.go

@@ -16,20 +16,20 @@ type pollDriverRegistration struct {
 }
 
 var pollDriverRegistrations = []pollDriverRegistration{
-	{driver: "nws_observation", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewObservationSource(cfg) }},
+	{driver: nws.DriverObservation, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewObservationSource(cfg) }},
-	{driver: "nws_alerts", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewAlertsSource(cfg) }},
+	{driver: nws.DriverAlerts, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewAlertsSource(cfg) }},
-	{driver: "nws_forecast_hourly", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewHourlyForecastSource(cfg) }},
+	{driver: nws.DriverForecastHourly, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewHourlyForecastSource(cfg) }},
-	{driver: "nws_forecast_narrative", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewNarrativeForecastSource(cfg) }},
+	{driver: nws.DriverForecastNarrative, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewNarrativeForecastSource(cfg) }},
-	{driver: "nws_forecast_discussion", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) {
+	{driver: nws.DriverForecastDiscussion, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) {
 		return nws.NewForecastDiscussionSource(cfg)
 	}},
-	{driver: "nws_weatherstories", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewWeatherStoriesSource(cfg) }},
+	{driver: nws.DriverWeatherStories, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewWeatherStoriesSource(cfg) }},
-	{driver: "openmeteo_observation", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return openmeteo.NewObservationSource(cfg) }},
+	{driver: openmeteo.DriverObservation, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return openmeteo.NewObservationSource(cfg) }},
-	{driver: "openmeteo_forecast", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return openmeteo.NewForecastSource(cfg) }},
+	{driver: openmeteo.DriverForecast, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return openmeteo.NewForecastSource(cfg) }},
-	{driver: "openweather_observation", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) {
+	{driver: openweather.DriverObservation, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) {
 		return openweather.NewObservationSource(cfg)
 	}},
-	{driver: "spc_convective_outlook", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) {
+	{driver: spc.DriverConvectiveOutlook, factory: func(cfg config.SourceConfig) (fksource.PollSource, error) {
 		return spc.NewConvectiveOutlookSource(cfg)
 	}},
 }

internal/sources/builtins_test.go

@@ -6,57 +6,29 @@ import (
 
 	"gitea.maximumdirect.net/ejr/feedkit/config"
 	fksource "gitea.maximumdirect.net/ejr/feedkit/sources"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/openweather"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/spc"
 )
 
-func TestRegisterBuiltinsRegistersNWSHourlyForecastDriver(t *testing.T) {
+func TestRegisterBuiltinsRegistersCurrentPollDrivers(t *testing.T) {
 	reg := fksource.NewRegistry()
 	RegisterBuiltins(reg)
 
-	in, err := reg.BuildInput(sourceConfigForDriver("nws_forecast_hourly"))
+	if len(pollDriverRegistrations) == 0 {
+		t.Fatalf("pollDriverRegistrations is empty")
+	}
+
+	for _, tt := range pollDriverRegistrations {
+		tt := tt
+		t.Run(tt.driver, func(t *testing.T) {
+			in, err := reg.BuildInput(sourceConfigForDriver(tt.driver))
 			if err != nil {
-		t.Fatalf("BuildInput(nws_forecast_hourly) error = %v", err)
+				t.Fatalf("BuildInput(%q) error = %v", tt.driver, err)
 			}
 			if _, ok := in.(fksource.PollSource); !ok {
-		t.Fatalf("BuildInput(nws_forecast_hourly) type = %T, want PollSource", in)
+				t.Fatalf("BuildInput(%q) type = %T, want PollSource", tt.driver, in)
 			}
-}
+		})
-
-func TestRegisterBuiltinsRegistersNWSNarrativeForecastDriver(t *testing.T) {
-	reg := fksource.NewRegistry()
-	RegisterBuiltins(reg)
-
-	in, err := reg.BuildInput(sourceConfigForDriver("nws_forecast_narrative"))
-	if err != nil {
-		t.Fatalf("BuildInput(nws_forecast_narrative) error = %v", err)
-	}
-	if _, ok := in.(fksource.PollSource); !ok {
-		t.Fatalf("BuildInput(nws_forecast_narrative) type = %T, want PollSource", in)
-	}
-}
-
-func TestRegisterBuiltinsRegistersNWSForecastDiscussionDriver(t *testing.T) {
-	reg := fksource.NewRegistry()
-	RegisterBuiltins(reg)
-
-	in, err := reg.BuildInput(sourceConfigForDriver("nws_forecast_discussion"))
-	if err != nil {
-		t.Fatalf("BuildInput(nws_forecast_discussion) error = %v", err)
-	}
-	if _, ok := in.(fksource.PollSource); !ok {
-		t.Fatalf("BuildInput(nws_forecast_discussion) type = %T, want PollSource", in)
-	}
-}
-
-func TestRegisterBuiltinsRegistersNWSWeatherStoriesDriver(t *testing.T) {
-	reg := fksource.NewRegistry()
-	RegisterBuiltins(reg)
-
-	in, err := reg.BuildInput(sourceConfigForDriver("nws_weatherstories"))
-	if err != nil {
-		t.Fatalf("BuildInput(nws_weatherstories) error = %v", err)
-	}
-	if _, ok := in.(fksource.PollSource); !ok {
-		t.Fatalf("BuildInput(nws_weatherstories) type = %T, want PollSource", in)
 	}
 }
 
@@ -73,44 +45,16 @@ func TestRegisterBuiltinsDoesNotRegisterLegacyNWSForecastDriver(t *testing.T) {
 	}
 }
 
-func TestRegisterBuiltinsRegistersAllCurrentDrivers(t *testing.T) {
-	reg := fksource.NewRegistry()
-	RegisterBuiltins(reg)
-
-	drivers := []string{
-		"nws_observation",
-		"nws_alerts",
-		"nws_forecast_hourly",
-		"nws_forecast_narrative",
-		"nws_forecast_discussion",
-		"nws_weatherstories",
-		"openmeteo_observation",
-		"openmeteo_forecast",
-		"openweather_observation",
-		"spc_convective_outlook",
-	}
-
-	for _, driver := range drivers {
-		in, err := reg.BuildInput(sourceConfigForDriver(driver))
-		if err != nil {
-			t.Fatalf("BuildInput(%s) error = %v", driver, err)
-		}
-		if _, ok := in.(fksource.PollSource); !ok {
-			t.Fatalf("BuildInput(%s) type = %T, want PollSource", driver, in)
-		}
-	}
-}
-
 func sourceConfigForDriver(driver string) config.SourceConfig {
 	url := "https://example.invalid"
-	if driver == "openweather_observation" {
+	if driver == openweather.DriverObservation {
 		url = "https://example.invalid?units=metric"
 	}
 	params := map[string]any{
 		"url":        url,
 		"user_agent": "test-agent",
 	}
-	if driver == "spc_convective_outlook" {
+	if driver == spc.DriverConvectiveOutlook {
 		params["latitude"] = 38.6239
 		params["longitude"] = -90.3571
 	}

internal/sources/docs_test.go

@@ -0,0 +1,32 @@
+package sources
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestDocumentedRegisteredSourceDrivers(t *testing.T) {
+	docs := map[string]string{
+		"docs/config.md":           readDoc(t, "../../docs/config.md"),
+		"docs/internal/sources.md": readDoc(t, "../../docs/internal/sources.md"),
+	}
+
+	for _, reg := range pollDriverRegistrations {
+		for path, doc := range docs {
+			if !strings.Contains(doc, reg.driver) {
+				t.Fatalf("%s missing source driver %q", path, reg.driver)
+			}
+		}
+	}
+}
+
+func readDoc(t *testing.T, path string) string {
+	t.Helper()
+
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("ReadFile(%s) error = %v", path, err)
+	}
+	return string(raw)
+}

internal/sources/internal/httpconfig/config.go

@@ -0,0 +1,58 @@
+package httpconfig
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+	"gitea.maximumdirect.net/ejr/feedkit/transport"
+)
+
+// Settings contains common HTTP client config for sources that fetch multiple documents.
+type Settings struct {
+	Name           string
+	UserAgent      string
+	Timeout        time.Duration
+	BodyLimitBytes int64
+}
+
+func Parse(driver string, cfg config.SourceConfig) (Settings, error) {
+	name := strings.TrimSpace(cfg.Name)
+	if name == "" {
+		return Settings{}, fmt.Errorf("%s: name is required", driver)
+	}
+	if cfg.Params == nil {
+		return Settings{}, fmt.Errorf("%s %q: params are required", driver, name)
+	}
+
+	userAgent, ok := cfg.ParamString("user_agent", "userAgent")
+	if !ok {
+		return Settings{}, fmt.Errorf("%s %q: params.user_agent is required", driver, name)
+	}
+
+	timeout := transport.DefaultHTTPTimeout
+	if _, exists := cfg.Params["http_timeout"]; exists {
+		var ok bool
+		timeout, ok = cfg.ParamDuration("http_timeout")
+		if !ok || timeout <= 0 {
+			return Settings{}, fmt.Errorf("source %q: params.http_timeout must be a positive duration", name)
+		}
+	}
+
+	bodyLimit := transport.DefaultHTTPResponseBodyLimitBytes
+	if _, exists := cfg.Params["http_response_body_limit_bytes"]; exists {
+		rawLimit, ok := cfg.ParamInt("http_response_body_limit_bytes")
+		if !ok || rawLimit <= 0 {
+			return Settings{}, fmt.Errorf("source %q: params.http_response_body_limit_bytes must be a positive integer", name)
+		}
+		bodyLimit = int64(rawLimit)
+	}
+
+	return Settings{
+		Name:           name,
+		UserAgent:      userAgent,
+		Timeout:        timeout,
+		BodyLimitBytes: bodyLimit,
+	}, nil
+}

internal/sources/internal/httpconfig/config_test.go

@@ -0,0 +1,115 @@
+package httpconfig
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+	"gitea.maximumdirect.net/ejr/feedkit/transport"
+)
+
+func TestParseUsesRequiredValuesAndDefaults(t *testing.T) {
+	got, err := Parse("test_driver", config.SourceConfig{
+		Name: " test-source ",
+		Params: map[string]any{
+			"user_agent": "test-agent",
+		},
+	})
+	if err != nil {
+		t.Fatalf("Parse() error = %v", err)
+	}
+	if got.Name != "test-source" {
+		t.Fatalf("Name = %q, want test-source", got.Name)
+	}
+	if got.UserAgent != "test-agent" {
+		t.Fatalf("UserAgent = %q, want test-agent", got.UserAgent)
+	}
+	if got.Timeout != transport.DefaultHTTPTimeout {
+		t.Fatalf("Timeout = %s, want %s", got.Timeout, transport.DefaultHTTPTimeout)
+	}
+	if got.BodyLimitBytes != transport.DefaultHTTPResponseBodyLimitBytes {
+		t.Fatalf("BodyLimitBytes = %d, want %d", got.BodyLimitBytes, transport.DefaultHTTPResponseBodyLimitBytes)
+	}
+}
+
+func TestParseUsesAliasesAndOverrides(t *testing.T) {
+	got, err := Parse("test_driver", config.SourceConfig{
+		Name: "test-source",
+		Params: map[string]any{
+			"userAgent":                      "test-agent",
+			"http_timeout":                   "2s",
+			"http_response_body_limit_bytes": 2048,
+		},
+	})
+	if err != nil {
+		t.Fatalf("Parse() error = %v", err)
+	}
+	if got.UserAgent != "test-agent" {
+		t.Fatalf("UserAgent = %q, want test-agent", got.UserAgent)
+	}
+	if got.Timeout != 2*time.Second {
+		t.Fatalf("Timeout = %s, want 2s", got.Timeout)
+	}
+	if got.BodyLimitBytes != 2048 {
+		t.Fatalf("BodyLimitBytes = %d, want 2048", got.BodyLimitBytes)
+	}
+}
+
+func TestParseRejectsInvalidConfig(t *testing.T) {
+	tests := []struct {
+		name    string
+		cfg     config.SourceConfig
+		wantErr string
+	}{
+		{
+			name:    "missing name",
+			cfg:     config.SourceConfig{Params: map[string]any{"user_agent": "test-agent"}},
+			wantErr: "test_driver: name is required",
+		},
+		{
+			name:    "missing params",
+			cfg:     config.SourceConfig{Name: "test-source"},
+			wantErr: `test_driver "test-source": params are required`,
+		},
+		{
+			name:    "missing user agent",
+			cfg:     config.SourceConfig{Name: "test-source", Params: map[string]any{}},
+			wantErr: `test_driver "test-source": params.user_agent is required`,
+		},
+		{
+			name: "invalid timeout",
+			cfg: config.SourceConfig{
+				Name: "test-source",
+				Params: map[string]any{
+					"user_agent":   "test-agent",
+					"http_timeout": "0s",
+				},
+			},
+			wantErr: `source "test-source": params.http_timeout must be a positive duration`,
+		},
+		{
+			name: "invalid body limit",
+			cfg: config.SourceConfig{
+				Name: "test-source",
+				Params: map[string]any{
+					"user_agent":                     "test-agent",
+					"http_response_body_limit_bytes": 0,
+				},
+			},
+			wantErr: `source "test-source": params.http_response_body_limit_bytes must be a positive integer`,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := Parse("test_driver", tt.cfg)
+			if err == nil {
+				t.Fatalf("Parse() error = nil, want %q", tt.wantErr)
+			}
+			if !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("Parse() error = %q, want %q", err, tt.wantErr)
+			}
+		})
+	}
+}

internal/sources/nws/alerts.go

@@ -26,10 +26,8 @@ type AlertsSource struct {
 }
 
 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 := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
+	hs, err := fksources.NewHTTPSource(DriverAlerts, cfg, "application/geo+json, application/json")
 	if err != nil {
 		return nil, err
 	}
@@ -40,7 +38,7 @@ func NewAlertsSource(cfg config.SourceConfig) (*AlertsSource, error) {
 func (s *AlertsSource) Name() string { return s.http.Name }
 
 // Kinds is used for routing/policy.
-func (s *AlertsSource) Kinds() []event.Kind { return []event.Kind{event.Kind("alert")} }
+func (s *AlertsSource) Kinds() []event.Kind { return []event.Kind{event.Kind(standards.KindAlert)} }
 
 func (s *AlertsSource) Poll(ctx context.Context) ([]event.Event, error) {
 	raw, meta, changed, err := s.fetchRaw(ctx)
@@ -71,7 +69,7 @@ func (s *AlertsSource) Poll(ctx context.Context) ([]event.Event, error) {
 	eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("alert"),
+		event.Kind(standards.KindAlert),
 		s.http.Name,
 		standards.SchemaRawNWSAlertsV1,
 		eventID,

internal/sources/nws/driver.go

@@ -0,0 +1,11 @@
+package nws
+
+// Source driver strings registered by weatherfeeder for NWS sources.
+const (
+	DriverObservation        = "nws_observation"
+	DriverAlerts             = "nws_alerts"
+	DriverForecastHourly     = "nws_forecast_hourly"
+	DriverForecastNarrative  = "nws_forecast_narrative"
+	DriverForecastDiscussion = "nws_forecast_discussion"
+	DriverWeatherStories     = "nws_weatherstories"
+)

internal/sources/nws/forecast_common.go

@@ -10,6 +10,7 @@ import (
 	"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"
 )
 
 const nwsForecastAccept = "application/geo+json, application/json"
@@ -44,7 +45,9 @@ func newForecastSource(cfg config.SourceConfig, driver, rawSchema string) (*fore
 
 func (s *forecastSource) Name() string { return s.http.Name }
 
-func (s *forecastSource) Kinds() []event.Kind { return []event.Kind{event.Kind("forecast")} }
+func (s *forecastSource) Kinds() []event.Kind {
+	return []event.Kind{event.Kind(standards.KindForecast)}
+}
 
 func (s *forecastSource) Poll(ctx context.Context) ([]event.Event, error) {
 	raw, meta, changed, err := s.fetchRaw(ctx)
@@ -69,7 +72,7 @@ func (s *forecastSource) Poll(ctx context.Context) ([]event.Event, error) {
 	eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("forecast"),
+		event.Kind(standards.KindForecast),
 		s.http.Name,
 		s.rawSchema,
 		eventID,

internal/sources/nws/forecast_discussion.go

@@ -20,9 +20,7 @@ type ForecastDiscussionSource struct {
 }
 
 func NewForecastDiscussionSource(cfg config.SourceConfig) (*ForecastDiscussionSource, error) {
-	const driver = "nws_forecast_discussion"
+	hs, err := fksources.NewHTTPSource(DriverForecastDiscussion, cfg, "text/html, application/xhtml+xml")
-
-	hs, err := fksources.NewHTTPSource(driver, cfg, "text/html, application/xhtml+xml")
 	if err != nil {
 		return nil, err
 	}
@@ -33,7 +31,7 @@ func NewForecastDiscussionSource(cfg config.SourceConfig) (*ForecastDiscussionSo
 func (s *ForecastDiscussionSource) Name() string { return s.http.Name }
 
 func (s *ForecastDiscussionSource) Kinds() []event.Kind {
-	return []event.Kind{event.Kind("forecast_discussion")}
+	return []event.Kind{event.Kind(standards.KindForecastDiscussion)}
 }
 
 func (s *ForecastDiscussionSource) Poll(ctx context.Context) ([]event.Event, error) {
@@ -57,7 +55,7 @@ func (s *ForecastDiscussionSource) Poll(ctx context.Context) ([]event.Event, err
 	eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("forecast_discussion"),
+		event.Kind(standards.KindForecastDiscussion),
 		s.http.Name,
 		standards.SchemaRawNWSForecastDiscussionV1,
 		eventID,

internal/sources/nws/forecast_discussion_test.go

@@ -27,7 +27,7 @@ func TestForecastDiscussionSourcePollEmitsExpectedEvent(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewForecastDiscussionSource() error = %v", err)
 	}
-	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind("forecast_discussion") {
+	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind(standards.KindForecastDiscussion) {
 		t.Fatalf("Kinds() = %#v, want [forecast_discussion]", got)
 	}
 
@@ -40,7 +40,7 @@ func TestForecastDiscussionSourcePollEmitsExpectedEvent(t *testing.T) {
 	}
 
 	got := events[0]
-	if got.Kind != event.Kind("forecast_discussion") {
+	if got.Kind != event.Kind(standards.KindForecastDiscussion) {
 		t.Fatalf("Kind = %q, want forecast_discussion", got.Kind)
 	}
 	if got.Schema != standards.SchemaRawNWSForecastDiscussionV1 {
@@ -117,7 +117,7 @@ func TestForecastDiscussionSourcePollRejectsInvalidHTML(t *testing.T) {
 func forecastDiscussionSourceConfig(url string) config.SourceConfig {
 	return config.SourceConfig{
 		Name:   "test-forecast-discussion-source",
-		Driver: "nws_forecast_discussion",
+		Driver: DriverForecastDiscussion,
 		Mode:   config.SourceModePoll,
 		Params: map[string]any{
 			"url":        url,

internal/sources/nws/forecast_hourly.go

@@ -18,8 +18,7 @@ type HourlyForecastSource struct {
 }
 
 func NewHourlyForecastSource(cfg config.SourceConfig) (*HourlyForecastSource, error) {
-	const driver = "nws_forecast_hourly"
+	src, err := newForecastSource(cfg, DriverForecastHourly, standards.SchemaRawNWSHourlyForecastV1)
-	src, err := newForecastSource(cfg, driver, standards.SchemaRawNWSHourlyForecastV1)
 	if err != nil {
 		return nil, err
 	}

internal/sources/nws/forecast_narrative.go

@@ -18,8 +18,7 @@ type NarrativeForecastSource struct {
 }
 
 func NewNarrativeForecastSource(cfg config.SourceConfig) (*NarrativeForecastSource, error) {
-	const driver = "nws_forecast_narrative"
+	src, err := newForecastSource(cfg, DriverForecastNarrative, standards.SchemaRawNWSNarrativeForecastV1)
-	src, err := newForecastSource(cfg, driver, standards.SchemaRawNWSNarrativeForecastV1)
 	if err != nil {
 		return nil, err
 	}

internal/sources/nws/forecast_test.go

@@ -26,7 +26,7 @@ func TestForecastSourcesEmitExpectedSchemaAndPreferGeneratedAt(t *testing.T) {
 	}{
 		{
 			name:       "hourly",
-			driver:     "nws_forecast_hourly",
+			driver:     DriverForecastHourly,
 			wantSchema: standards.SchemaRawNWSHourlyForecastV1,
 			newSource: func(cfg config.SourceConfig) (forecastPoller, error) {
 				return NewHourlyForecastSource(cfg)
@@ -34,7 +34,7 @@ func TestForecastSourcesEmitExpectedSchemaAndPreferGeneratedAt(t *testing.T) {
 		},
 		{
 			name:       "narrative",
-			driver:     "nws_forecast_narrative",
+			driver:     DriverForecastNarrative,
 			wantSchema: standards.SchemaRawNWSNarrativeForecastV1,
 			newSource: func(cfg config.SourceConfig) (forecastPoller, error) {
 				return NewNarrativeForecastSource(cfg)
@@ -55,7 +55,7 @@ func TestForecastSourcesEmitExpectedSchemaAndPreferGeneratedAt(t *testing.T) {
 			}
 			if ks, ok := src.(interface{ Kinds() []event.Kind }); !ok {
 				t.Fatalf("source does not implement Kinds()")
-			} else if gotKinds := ks.Kinds(); len(gotKinds) != 1 || gotKinds[0] != event.Kind("forecast") {
+			} else if gotKinds := ks.Kinds(); len(gotKinds) != 1 || gotKinds[0] != event.Kind(standards.KindForecast) {
 				t.Fatalf("Kinds() = %#v, want [forecast]", gotKinds)
 			}
 
@@ -69,7 +69,7 @@ func TestForecastSourcesEmitExpectedSchemaAndPreferGeneratedAt(t *testing.T) {
 			if got[0].Schema != tt.wantSchema {
 				t.Fatalf("Poll() schema = %q, want %q", got[0].Schema, tt.wantSchema)
 			}
-			if got[0].Kind != event.Kind("forecast") {
+			if got[0].Kind != event.Kind(standards.KindForecast) {
 				t.Fatalf("Poll() kind = %q, want forecast", got[0].Kind)
 			}
 
@@ -117,7 +117,7 @@ func TestForecastSourcePollEffectiveAtFallbackOrder(t *testing.T) {
 			}))
 			defer srv.Close()
 
-			src, err := NewHourlyForecastSource(forecastSourceConfig("nws_forecast_hourly", srv.URL))
+			src, err := NewHourlyForecastSource(forecastSourceConfig(DriverForecastHourly, srv.URL))
 			if err != nil {
 				t.Fatalf("NewHourlyForecastSource() error = %v", err)
 			}
@@ -148,7 +148,7 @@ func TestForecastSourcePollMetadataDecodeFailureStillEmitsRawEvent(t *testing.T)
 	}))
 	defer srv.Close()
 
-	src, err := NewNarrativeForecastSource(forecastSourceConfig("nws_forecast_narrative", srv.URL))
+	src, err := NewNarrativeForecastSource(forecastSourceConfig(DriverForecastNarrative, srv.URL))
 	if err != nil {
 		t.Fatalf("NewNarrativeForecastSource() error = %v", err)
 	}

internal/sources/nws/observation.go

@@ -20,9 +20,7 @@ type ObservationSource struct {
 }
 
 func NewObservationSource(cfg config.SourceConfig) (*ObservationSource, error) {
-	const driver = "nws_observation"
+	hs, err := fksources.NewHTTPSource(DriverObservation, cfg, "application/geo+json, application/json")
-
-	hs, err := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
 	if err != nil {
 		return nil, err
 	}
@@ -32,7 +30,9 @@ func NewObservationSource(cfg config.SourceConfig) (*ObservationSource, error) {
 
 func (s *ObservationSource) Name() string { return s.http.Name }
 
-func (s *ObservationSource) Kinds() []event.Kind { return []event.Kind{event.Kind("observation")} }
+func (s *ObservationSource) Kinds() []event.Kind {
+	return []event.Kind{event.Kind(standards.KindObservation)}
+}
 
 func (s *ObservationSource) Poll(ctx context.Context) ([]event.Event, error) {
 	raw, meta, changed, err := s.fetchRaw(ctx)
@@ -54,7 +54,7 @@ func (s *ObservationSource) Poll(ctx context.Context) ([]event.Event, error) {
 	eventID := fksources.DefaultEventID(meta.ID, s.http.Name, effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("observation"),
+		event.Kind(standards.KindObservation),
 		s.http.Name,
 		standards.SchemaRawNWSObservationV1,
 		eventID,

internal/sources/nws/observation_test.go

@@ -8,6 +8,7 @@ import (
 
 	"gitea.maximumdirect.net/ejr/feedkit/config"
 	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
 )
 
 func TestObservationSourcePollReturnsNoEventsOn304(t *testing.T) {
@@ -31,7 +32,7 @@ func TestObservationSourcePollReturnsNoEventsOn304(t *testing.T) {
 
 	src, err := NewObservationSource(config.SourceConfig{
 		Name:   "NWSObservationTest",
-		Driver: "nws_observation",
+		Driver: DriverObservation,
 		Mode:   config.SourceModePoll,
 		Params: map[string]any{
 			"url":        srv.URL,
@@ -41,7 +42,7 @@ func TestObservationSourcePollReturnsNoEventsOn304(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewObservationSource() error = %v", err)
 	}
-	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind("observation") {
+	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind(standards.KindObservation) {
 		t.Fatalf("Kinds() = %#v, want [observation]", got)
 	}
 
@@ -52,7 +53,7 @@ func TestObservationSourcePollReturnsNoEventsOn304(t *testing.T) {
 	if len(first) != 1 {
 		t.Fatalf("first Poll() len = %d, want 1", len(first))
 	}
-	if first[0].Kind != event.Kind("observation") {
+	if first[0].Kind != event.Kind(standards.KindObservation) {
 		t.Fatalf("first Poll() kind = %q", first[0].Kind)
 	}
 

internal/sources/nws/weatherstories.go

@@ -22,9 +22,7 @@ type WeatherStoriesSource struct {
 }
 
 func NewWeatherStoriesSource(cfg config.SourceConfig) (*WeatherStoriesSource, error) {
-	const driver = "nws_weatherstories"
+	hs, err := fksources.NewHTTPSource(DriverWeatherStories, cfg, "application/geo+json, application/json")
-
-	hs, err := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json")
 	if err != nil {
 		return nil, err
 	}
@@ -35,7 +33,7 @@ func NewWeatherStoriesSource(cfg config.SourceConfig) (*WeatherStoriesSource, er
 func (s *WeatherStoriesSource) Name() string { return s.http.Name }
 
 func (s *WeatherStoriesSource) Kinds() []event.Kind {
-	return []event.Kind{event.Kind("weather_story")}
+	return []event.Kind{event.Kind(standards.KindWeatherStory)}
 }
 
 func (s *WeatherStoriesSource) Poll(ctx context.Context) ([]event.Event, error) {
@@ -61,7 +59,7 @@ func (s *WeatherStoriesSource) Poll(ctx context.Context) ([]event.Event, error)
 	eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("weather_story"),
+		event.Kind(standards.KindWeatherStory),
 		s.http.Name,
 		standards.SchemaRawNWSWeatherStoriesV1,
 		eventID,

internal/sources/nws/weatherstories_test.go

@@ -28,7 +28,7 @@ func TestWeatherStoriesSourcePollEmitsExpectedEventAndPrefersLatestUpdateTime(t
 	if err != nil {
 		t.Fatalf("NewWeatherStoriesSource() error = %v", err)
 	}
-	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind("weather_story") {
+	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind(standards.KindWeatherStory) {
 		t.Fatalf("Kinds() = %#v, want [weather_story]", got)
 	}
 
@@ -41,7 +41,7 @@ func TestWeatherStoriesSourcePollEmitsExpectedEventAndPrefersLatestUpdateTime(t
 	}
 
 	got := events[0]
-	if got.Kind != event.Kind("weather_story") {
+	if got.Kind != event.Kind(standards.KindWeatherStory) {
 		t.Fatalf("Kind = %q, want weather_story", got.Kind)
 	}
 	if got.Schema != standards.SchemaRawNWSWeatherStoriesV1 {
@@ -148,7 +148,7 @@ func TestWeatherStoriesSourcePollMetadataDecodeFailureStillEmitsRawEvent(t *test
 func weatherStoriesSourceConfig(url string) config.SourceConfig {
 	return config.SourceConfig{
 		Name:   "test-weatherstories-source",
-		Driver: "nws_weatherstories",
+		Driver: DriverWeatherStories,
 		Mode:   config.SourceModePoll,
 		Params: map[string]any{
 			"url":        url,

internal/sources/openmeteo/driver.go

@@ -0,0 +1,7 @@
+package openmeteo
+
+// Source driver strings registered by weatherfeeder for Open-Meteo sources.
+const (
+	DriverObservation = "openmeteo_observation"
+	DriverForecast    = "openmeteo_forecast"
+)

internal/sources/openmeteo/forecast.go

@@ -19,9 +19,7 @@ type ForecastSource struct {
 }
 
 func NewForecastSource(cfg config.SourceConfig) (*ForecastSource, error) {
-	const driver = "openmeteo_forecast"
+	hs, err := fksources.NewHTTPSource(DriverForecast, cfg, "application/json")
-
-	hs, err := fksources.NewHTTPSource(driver, cfg, "application/json")
 	if err != nil {
 		return nil, err
 	}
@@ -31,7 +29,9 @@ func NewForecastSource(cfg config.SourceConfig) (*ForecastSource, error) {
 
 func (s *ForecastSource) Name() string { return s.http.Name }
 
-func (s *ForecastSource) Kinds() []event.Kind { return []event.Kind{event.Kind("forecast")} }
+func (s *ForecastSource) Kinds() []event.Kind {
+	return []event.Kind{event.Kind(standards.KindForecast)}
+}
 
 func (s *ForecastSource) Poll(ctx context.Context) ([]event.Event, error) {
 	raw, meta, changed, err := s.fetchRaw(ctx)
@@ -55,7 +55,7 @@ func (s *ForecastSource) Poll(ctx context.Context) ([]event.Event, error) {
 	eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("forecast"),
+		event.Kind(standards.KindForecast),
 		s.http.Name,
 		standards.SchemaRawOpenMeteoHourlyForecastV1,
 		eventID,

internal/sources/openmeteo/observation.go

@@ -19,9 +19,7 @@ type ObservationSource struct {
 }
 
 func NewObservationSource(cfg config.SourceConfig) (*ObservationSource, error) {
-	const driver = "openmeteo_observation"
+	hs, err := fksources.NewHTTPSource(DriverObservation, cfg, "application/json")
-
-	hs, err := fksources.NewHTTPSource(driver, cfg, "application/json")
 	if err != nil {
 		return nil, err
 	}
@@ -31,7 +29,9 @@ func NewObservationSource(cfg config.SourceConfig) (*ObservationSource, error) {
 
 func (s *ObservationSource) Name() string { return s.http.Name }
 
-func (s *ObservationSource) Kinds() []event.Kind { return []event.Kind{event.Kind("observation")} }
+func (s *ObservationSource) Kinds() []event.Kind {
+	return []event.Kind{event.Kind(standards.KindObservation)}
+}
 
 func (s *ObservationSource) Poll(ctx context.Context) ([]event.Event, error) {
 	raw, meta, changed, err := s.fetchRaw(ctx)
@@ -52,7 +52,7 @@ func (s *ObservationSource) Poll(ctx context.Context) ([]event.Event, error) {
 	eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("observation"),
+		event.Kind(standards.KindObservation),
 		s.http.Name,
 		standards.SchemaRawOpenMeteoCurrentV1,
 		eventID,

internal/sources/openmeteo/source_test.go

@@ -5,12 +5,13 @@ import (
 
 	"gitea.maximumdirect.net/ejr/feedkit/config"
 	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
 )
 
 func TestObservationSourceAdvertisesKinds(t *testing.T) {
 	src, err := NewObservationSource(config.SourceConfig{
 		Name:   "openmeteo-observation-test",
-		Driver: "openmeteo_observation",
+		Driver: DriverObservation,
 		Mode:   config.SourceModePoll,
 		Params: map[string]any{
 			"url":        "https://example.invalid",
@@ -20,7 +21,7 @@ func TestObservationSourceAdvertisesKinds(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewObservationSource() error = %v", err)
 	}
-	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind("observation") {
+	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind(standards.KindObservation) {
 		t.Fatalf("Kinds() = %#v, want [observation]", got)
 	}
 }
@@ -28,7 +29,7 @@ func TestObservationSourceAdvertisesKinds(t *testing.T) {
 func TestForecastSourceAdvertisesKinds(t *testing.T) {
 	src, err := NewForecastSource(config.SourceConfig{
 		Name:   "openmeteo-forecast-test",
-		Driver: "openmeteo_forecast",
+		Driver: DriverForecast,
 		Mode:   config.SourceModePoll,
 		Params: map[string]any{
 			"url":        "https://example.invalid",
@@ -38,7 +39,7 @@ func TestForecastSourceAdvertisesKinds(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewForecastSource() error = %v", err)
 	}
-	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind("forecast") {
+	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind(standards.KindForecast) {
 		t.Fatalf("Kinds() = %#v, want [forecast]", got)
 	}
 }

internal/sources/openweather/driver.go

@@ -0,0 +1,6 @@
+package openweather
+
+// Source driver strings registered by weatherfeeder for OpenWeather sources.
+const (
+	DriverObservation = "openweather_observation"
+)

internal/sources/openweather/observation.go

@@ -19,9 +19,7 @@ type ObservationSource struct {
 }
 
 func NewObservationSource(cfg config.SourceConfig) (*ObservationSource, error) {
-	const driver = "openweather_observation"
+	hs, err := fksources.NewHTTPSource(DriverObservation, cfg, "application/json")
-
-	hs, err := fksources.NewHTTPSource(driver, cfg, "application/json")
 	if err != nil {
 		return nil, err
 	}
@@ -35,7 +33,9 @@ func NewObservationSource(cfg config.SourceConfig) (*ObservationSource, error) {
 
 func (s *ObservationSource) Name() string { return s.http.Name }
 
-func (s *ObservationSource) Kinds() []event.Kind { return []event.Kind{event.Kind("observation")} }
+func (s *ObservationSource) Kinds() []event.Kind {
+	return []event.Kind{event.Kind(standards.KindObservation)}
+}
 
 func (s *ObservationSource) Poll(ctx context.Context) ([]event.Event, error) {
 	if err := owcommon.RequireMetricUnits(s.http.URL); err != nil {
@@ -60,7 +60,7 @@ func (s *ObservationSource) Poll(ctx context.Context) ([]event.Event, error) {
 	eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("observation"),
+		event.Kind(standards.KindObservation),
 		s.http.Name,
 		standards.SchemaRawOpenWeatherCurrentV1,
 		eventID,

internal/sources/openweather/source_test.go

@@ -5,12 +5,13 @@ import (
 
 	"gitea.maximumdirect.net/ejr/feedkit/config"
 	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
 )
 
 func TestObservationSourceAdvertisesKinds(t *testing.T) {
 	src, err := NewObservationSource(config.SourceConfig{
 		Name:   "openweather-observation-test",
-		Driver: "openweather_observation",
+		Driver: DriverObservation,
 		Mode:   config.SourceModePoll,
 		Params: map[string]any{
 			"url":        "https://example.invalid?units=metric",
@@ -20,7 +21,7 @@ func TestObservationSourceAdvertisesKinds(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewObservationSource() error = %v", err)
 	}
-	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind("observation") {
+	if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind(standards.KindObservation) {
 		t.Fatalf("Kinds() = %#v, want [observation]", got)
 	}
 }

internal/sources/spc/convective_outlook.go

@@ -16,12 +16,11 @@ import (
 	fksources "gitea.maximumdirect.net/ejr/feedkit/sources"
 	"gitea.maximumdirect.net/ejr/feedkit/transport"
 	spcprovider "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/spc"
+	"gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/internal/httpconfig"
 	"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
 )
 
 const (
-	driverConvectiveOutlook = "spc_convective_outlook"
-
 	acceptGeoJSON    = "application/geo+json, application/json"
 	acceptDiscussion = "text/html, application/xhtml+xml"
 	acceptRSS        = "application/rss+xml, application/xml, text/xml"
@@ -56,53 +55,27 @@ type ConvectiveOutlookSource struct {
 }
 
 func NewConvectiveOutlookSource(cfg config.SourceConfig) (*ConvectiveOutlookSource, error) {
-	name := strings.TrimSpace(cfg.Name)
+	httpSettings, err := httpconfig.Parse(DriverConvectiveOutlook, cfg)
-	if name == "" {
+	if err != nil {
-		return nil, fmt.Errorf("%s: name is required", driverConvectiveOutlook)
+		return nil, err
-	}
-	if cfg.Params == nil {
-		return nil, fmt.Errorf("%s %q: params are required", driverConvectiveOutlook, name)
-	}
-
-	userAgent, ok := cfg.ParamString("user_agent", "userAgent")
-	if !ok {
-		return nil, fmt.Errorf("%s %q: params.user_agent is required", driverConvectiveOutlook, name)
 	}
 
 	latitude, err := requireFloatParam(cfg, "latitude")
 	if err != nil {
-		return nil, fmt.Errorf("%s %q: %w", driverConvectiveOutlook, name, err)
+		return nil, fmt.Errorf("%s %q: %w", DriverConvectiveOutlook, httpSettings.Name, err)
 	}
 	longitude, err := requireFloatParam(cfg, "longitude")
 	if err != nil {
-		return nil, fmt.Errorf("%s %q: %w", driverConvectiveOutlook, name, err)
+		return nil, fmt.Errorf("%s %q: %w", DriverConvectiveOutlook, httpSettings.Name, err)
-	}
-
-	timeout := transport.DefaultHTTPTimeout
-	if _, exists := cfg.Params["http_timeout"]; exists {
-		var ok bool
-		timeout, ok = cfg.ParamDuration("http_timeout")
-		if !ok || timeout <= 0 {
-			return nil, fmt.Errorf("source %q: params.http_timeout must be a positive duration", name)
-		}
-	}
-
-	bodyLimit := transport.DefaultHTTPResponseBodyLimitBytes
-	if _, exists := cfg.Params["http_response_body_limit_bytes"]; exists {
-		rawLimit, ok := cfg.ParamInt("http_response_body_limit_bytes")
-		if !ok || rawLimit <= 0 {
-			return nil, fmt.Errorf("source %q: params.http_response_body_limit_bytes must be a positive integer", name)
-		}
-		bodyLimit = int64(rawLimit)
 	}
 
 	geoJSONProducts, err := configuredGeoJSONProducts(cfg)
 	if err != nil {
-		return nil, fmt.Errorf("%s %q: %w", driverConvectiveOutlook, name, err)
+		return nil, fmt.Errorf("%s %q: %w", DriverConvectiveOutlook, httpSettings.Name, err)
 	}
 	discussions, err := configuredDiscussionProducts(cfg)
 	if err != nil {
-		return nil, fmt.Errorf("%s %q: %w", driverConvectiveOutlook, name, err)
+		return nil, fmt.Errorf("%s %q: %w", DriverConvectiveOutlook, httpSettings.Name, err)
 	}
 
 	rssURL := ""
@@ -114,14 +87,14 @@ func NewConvectiveOutlookSource(cfg config.SourceConfig) (*ConvectiveOutlookSour
 	locationName, _ := cfg.ParamString("location_name", "locationName")
 
 	return &ConvectiveOutlookSource{
-		name:            name,
+		name:            httpSettings.Name,
-		userAgent:       userAgent,
+		userAgent:       httpSettings.UserAgent,
 		locationID:      locationID,
 		locationName:    locationName,
 		latitude:        latitude,
 		longitude:       longitude,
-		client:          transport.NewHTTPClient(timeout),
+		client:          transport.NewHTTPClient(httpSettings.Timeout),
-		bodyLimit:       bodyLimit,
+		bodyLimit:       httpSettings.BodyLimitBytes,
 		geoJSONProducts: geoJSONProducts,
 		discussions:     discussions,
 		rssURL:          rssURL,
@@ -131,7 +104,7 @@ func NewConvectiveOutlookSource(cfg config.SourceConfig) (*ConvectiveOutlookSour
 func (s *ConvectiveOutlookSource) Name() string { return s.name }
 
 func (s *ConvectiveOutlookSource) Kinds() []event.Kind {
-	return []event.Kind{event.Kind("outlook")}
+	return []event.Kind{event.Kind(standards.KindOutlook)}
 }
 
 func (s *ConvectiveOutlookSource) Poll(ctx context.Context) ([]event.Event, error) {
@@ -221,7 +194,7 @@ func (s *ConvectiveOutlookSource) Poll(ctx context.Context) ([]event.Event, erro
 	eventID := fksources.DefaultEventID("", s.name, &effectiveAt, emittedAt)
 
 	return fksources.SingleEvent(
-		event.Kind("outlook"),
+		event.Kind(standards.KindOutlook),
 		s.name,
 		standards.SchemaRawSPCConvectiveOutlookV1,
 		eventID,

internal/sources/spc/convective_outlook_test.go

@@ -4,8 +4,6 @@ import (
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
-	"os"
-	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -22,7 +20,7 @@ func TestConvectiveOutlookSourceKinds(t *testing.T) {
 		t.Fatalf("NewConvectiveOutlookSource() error = %v", err)
 	}
 	got := src.Kinds()
-	if len(got) != 1 || got[0] != event.Kind("outlook") {
+	if len(got) != 1 || got[0] != event.Kind(standards.KindOutlook) {
 		t.Fatalf("Kinds() = %#v, want [outlook]", got)
 	}
 }
@@ -57,7 +55,7 @@ func TestConvectiveOutlookSourcePollEmitsRawBundle(t *testing.T) {
 		t.Fatalf("Poll() returned %d events, want 1", len(events))
 	}
 	got := events[0]
-	if got.Kind != event.Kind("outlook") {
+	if got.Kind != event.Kind(standards.KindOutlook) {
 		t.Fatalf("Kind = %q, want outlook", got.Kind)
 	}
 	if got.Schema != standards.SchemaRawSPCConvectiveOutlookV1 {
@@ -78,8 +76,8 @@ func TestConvectiveOutlookSourcePollEmitsRawBundle(t *testing.T) {
 	if bundle.Latitude != 38.6239 || bundle.Longitude != -90.3571 {
 		t.Fatalf("coordinates = %v,%v", bundle.Latitude, bundle.Longitude)
 	}
-	if len(bundle.Products) != 12 {
+	if len(bundle.Products) != 9 {
-		t.Fatalf("Products length = %d, want 12", len(bundle.Products))
+		t.Fatalf("Products length = %d, want 9", len(bundle.Products))
 	}
 	if len(bundle.Discussions) != 3 {
 		t.Fatalf("Discussions length = %d, want 3", len(bundle.Discussions))
@@ -277,7 +275,7 @@ func convectiveOutlookConfig(extra map[string]any) config.SourceConfig {
 	}
 	return config.SourceConfig{
 		Name:   "spc-test",
-		Driver: driverConvectiveOutlook,
+		Driver: DriverConvectiveOutlook,
 		Mode:   config.SourceModePoll,
 		Params: params,
 	}
@@ -312,7 +310,7 @@ func geoJSONFixtureForProduct(t *testing.T, key string, blankIssueISO bool) []by
 	case strings.HasPrefix(key, "day2_"):
 		name = "day2_torn.geojson"
 	case strings.HasPrefix(key, "day3_"):
-		name = "day3_wind.geojson"
+		name = "day3_cat.geojson"
 	default:
 		t.Fatalf("unknown product key %q", key)
 	}
@@ -340,16 +338,6 @@ func discussionFixtureForProduct(t *testing.T, key string) []byte {
 	}
 }
 
-func readSPCTestFixture(t *testing.T, name string) []byte {
-	t.Helper()
-	path := filepath.Join("..", "..", "providers", "spc", "testdata", name)
-	raw, err := os.ReadFile(path)
-	if err != nil {
-		t.Fatalf("read fixture %s: %v", path, err)
-	}
-	return raw
-}
-
 const testRSS = `<?xml version="1.0"?>
 <rss version="2.0">
   <channel>

internal/sources/spc/driver.go

@@ -0,0 +1,6 @@
+package spc
+
+// Source driver strings registered by weatherfeeder for SPC sources.
+const (
+	DriverConvectiveOutlook = "spc_convective_outlook"
+)

internal/sources/spc/fixture_test.go

@@ -0,0 +1,17 @@
+package spc
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func readSPCTestFixture(t *testing.T, name string) []byte {
+	t.Helper()
+	path := filepath.Join("..", "..", "providers", "spc", "testdata", name)
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read fixture %s: %v", path, err)
+	}
+	return raw
+}

model/docs_test.go

@@ -0,0 +1,37 @@
+package model
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestDocumentedConsumerModelTypes(t *testing.T) {
+	raw, err := os.ReadFile("../docs/consumers/pkg-model.md")
+	if err != nil {
+		t.Fatalf("ReadFile(pkg-model.md) error = %v", err)
+	}
+	doc := string(raw)
+
+	required := []string{
+		"gitea.maximumdirect.net/ejr/weatherfeeder/model",
+		"WeatherObservation",
+		"WeatherForecastRun",
+		"WeatherForecastPeriod",
+		"WeatherForecastDiscussion",
+		"WeatherForecastDiscussionSection",
+		"WeatherStoryRun",
+		"WeatherStory",
+		"WeatherAlertRun",
+		"WeatherAlert",
+		"WeatherAlertReference",
+		"WeatherOutlookRun",
+		"WeatherOutlook",
+		"WMOCode",
+	}
+	for _, want := range required {
+		if !strings.Contains(doc, want) {
+			t.Fatalf("docs/consumers/pkg-model.md missing %q", want)
+		}
+	}
+}

standards/docs_test.go

@@ -0,0 +1,101 @@
+package standards
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+func TestDocumentedEventSchemas(t *testing.T) {
+	raw, err := os.ReadFile("../docs/integrations/events.md")
+	if err != nil {
+		t.Fatalf("ReadFile(events.md) error = %v", err)
+	}
+	doc := string(raw)
+
+	for _, schema := range schemaConstants(t, false) {
+		if !strings.Contains(doc, schema) {
+			t.Fatalf("docs/integrations/events.md missing schema %q", schema)
+		}
+	}
+}
+
+func TestDocumentedConsumerStandardsConstants(t *testing.T) {
+	raw, err := os.ReadFile("../docs/consumers/pkg-standards.md")
+	if err != nil {
+		t.Fatalf("ReadFile(pkg-standards.md) error = %v", err)
+	}
+	doc := string(raw)
+
+	for _, schema := range schemaConstants(t, true) {
+		if !strings.Contains(doc, schema) {
+			t.Fatalf("docs/consumers/pkg-standards.md missing schema %q", schema)
+		}
+	}
+	for _, kind := range kindConstants(t) {
+		if !strings.Contains(doc, kind) {
+			t.Fatalf("docs/consumers/pkg-standards.md missing kind %q", kind)
+		}
+	}
+}
+
+func schemaConstants(t *testing.T, includeNonCurrent bool) []string {
+	t.Helper()
+
+	return stringConstantsFromFile(t, "schema.go", "Schema", func(name string) bool {
+		return !includeNonCurrent && schemaConstantNotInCurrentContract(name)
+	})
+}
+
+func kindConstants(t *testing.T) []string {
+	t.Helper()
+
+	return stringConstantsFromFile(t, "kind.go", "Kind", nil)
+}
+
+func stringConstantsFromFile(t *testing.T, path string, prefix string, skip func(string) bool) []string {
+	t.Helper()
+
+	file, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
+	if err != nil {
+		t.Fatalf("ParseFile(%s) error = %v", path, err)
+	}
+
+	var out []string
+	ast.Inspect(file, func(n ast.Node) bool {
+		valueSpec, ok := n.(*ast.ValueSpec)
+		if !ok {
+			return true
+		}
+		for i, name := range valueSpec.Names {
+			if !strings.HasPrefix(name.Name, prefix) || (skip != nil && skip(name.Name)) {
+				continue
+			}
+			if i >= len(valueSpec.Values) {
+				t.Fatalf("constant %s has no explicit value", name.Name)
+			}
+			lit, ok := valueSpec.Values[i].(*ast.BasicLit)
+			if !ok || lit.Kind != token.STRING {
+				t.Fatalf("constant %s is not a string literal", name.Name)
+			}
+			value, err := strconv.Unquote(lit.Value)
+			if err != nil {
+				t.Fatalf("constant %s value is not a quoted string: %v", name.Name, err)
+			}
+			out = append(out, value)
+		}
+		return true
+	})
+	if len(out) == 0 {
+		t.Fatalf("no %s constants found in %s", prefix, path)
+	}
+	return out
+}
+
+func schemaConstantNotInCurrentContract(name string) bool {
+	return name == "SchemaRawOpenWeatherHourlyForecastV1"
+}

standards/kind.go

@@ -0,0 +1,11 @@
+package standards
+
+// Event kind strings used by weatherfeeder events and routing policy.
+const (
+	KindObservation        = "observation"
+	KindForecast           = "forecast"
+	KindForecastDiscussion = "forecast_discussion"
+	KindWeatherStory       = "weather_story"
+	KindAlert              = "alert"
+	KindOutlook            = "outlook"
+)