weatherfeeder/docs/internal/postgres-sink.md

Postgres Sink Internals

Purpose

internal/sinks/postgres defines weatherfeeder's canonical-event-to-Postgres mapping. It supplies a schema definition and mapper to feedkit's generic Postgres sink.

The consumer-facing table contract is docs/integrations/postgres.md. This document describes the internal ownership boundary.

Inputs And Outputs

Inputs are canonical feed events. The mapper currently handles these schemas:

• weather.observation.v1
• weather.forecast.v1
• weather.forecast_discussion.v1
• weather.weather_story.v1
• weather.alert.v1
• weather.outlook.v1

Outputs are feedkit PostgresWrite values for weatherfeeder-owned tables. Unsupported schemas produce no writes and no error.

Boundaries

• Weatherfeeder owns table definitions in schema.go.
• Weatherfeeder owns canonical payload mapping in map.go.
• Feedkit owns database opening, table and index creation, transactions, inserts, context-aware consumption, and prune execution.
• Postgres mapping consumes canonical events only. It should not understand raw provider schemas.

Config Fields Used

Weatherfeeder registers the postgres sink by passing PostgresSchema() to feedkit. Feedkit parses sink params:

• uri
• username
• password
• prune, optional duration

Weatherfeeder-owned mapper code does not read config directly.

External Adapters Used

The runtime registers the sink with:

sinkReg.Register("postgres", fksinks.PostgresFactory(wfpgsink.PostgresSchema()))

Feedkit validates events at the sink boundary, calls the weatherfeeder mapper, validates writes against the compiled schema, inserts rows in a transaction, and optionally prunes rows older than the configured window.

State

The mapper is stateless. Durable state is stored in Postgres through feedkit's sink implementation.

Mapping Rules

Parent rows preserve event envelope fields where the table supports them:

• event_id
• event_kind
• event_source
• event_schema
• event_emitted_at
• event_effective_at

Child rows use positional indexes to preserve canonical array order:

• weather_index
• period_index
• message_index
• story_index
• alert_index
• reference_index
• outlook_index

Required canonical fields are validated before writes are returned:

• observations require timestamp;
• forecasts require issuedAt and product, and each period requires startTime and endTime;
• forecast discussions require issuedAt and product;
• weather story runs require asOf, and each story requires startTime, endTime, and updatedAt;
• alert runs require asOf, and each alert requires id;
• outlook runs require asOf, and each outlook requires id, provider, product, day, outlookType, label, validFrom, validTo, issuedAt, expiresAt, and geometry.

Nullable canonical values are converted to SQL nulls by mapper helpers. Observation present-weather raw values and outlook geometry values are stored as compact JSON text.

Failure Behavior

Payload decode failures, missing required fields, invalid compact JSON values, or schema/write mismatches return errors to feedkit's sink. Feedkit rolls back the transaction when a write fails.

Unsupported canonical schemas are ignored by this mapper so other routed events can use different sinks without Postgres-specific failures.

Tests To Inspect

• internal/sinks/postgres/schema_test.go
• internal/sinks/postgres/map_test.go
• feedkit Postgres sink tests when changing generic sink behavior assumptions

Invariants

• Persist only canonical schemas.
• Preserve event envelope fields in parent rows.
• Preserve array order with child positional indexes.
• Validate required fields before writing.
• Keep table-contract docs synchronized with schema and mapper changes.