Moved broadly useful helper functions upstream into feedkit

README.md

@@ -15,12 +15,12 @@ lifecycle, domain schemas, and domain-specific validation in your daemon.
 
 ## Conceptual pipeline
 
-Collect -> Process (optional stages, including normalize) -> Route -> Emit
+Collect -> Process (optional stages, including dedupe + normalize) -> Route -> Emit
 
 | Stage | Package(s) |
 |---|---|
 | Collect | `sources`, `scheduler` |
-| Process | `pipeline`, `processors`, `normalize` (optional stage) |
+| Process | `pipeline`, `processors`, `processors/dedupe`, `processors/normalize` (optional stages) |
 | Route | `dispatch` |
 | Emit | `sinks` |
 | Configure | `config` |
@@ -86,7 +86,11 @@ Processors can transform, drop, or reject events.
 Defines the generic processor interface and a named-driver registry used by
 daemons to build ordered processor chains.
 
-### `normalize`
+### `processors/dedupe`
+
+Built-in in-memory LRU dedupe processor that drops repeated events by `Event.ID`.
+
+### `processors/normalize`
 
 Concrete normalization processor implementation. Typical use: sources emit raw
 payload events, then a normalize stage maps them to canonical schemas.
@@ -97,8 +101,13 @@ Compiles routes and fans out events to sinks with per-sink queue/worker isolatio
 
 ### `sinks`
 
-Defines sink interface and sink registry. Built-ins include `stdout` and `nats`, with
+Defines sink interface and sink registry. Built-ins include:
-additional sink implementations at varying maturity.
+- `stdout`
+- `nats`
+- `postgres`
+
+Detailed Postgres configuration and wiring examples live in package docs:
+`sinks/doc.go`.
 
 ## Typical wiring
 

doc.go

@@ -5,12 +5,12 @@
 //
 // Conceptual flow:
 //
-//	Collect -> Process (optional stages, including normalize) -> Route -> Emit
+//	Collect -> Process (optional stages, including dedupe + normalize) -> Route -> Emit
 //
 // In feedkit this maps to:
 //
 //	Collect: sources + scheduler
-//	Process: pipeline + processors + normalize (optional stage)
+//	Process: pipeline + processors + processors/dedupe + processors/normalize (optional stages)
 //	Route:   dispatch
 //	Emit:    sinks
 //	Config:  config
@@ -50,6 +50,19 @@
 //     Both share Input{Name()}. A source may emit 0..N events per poll/run step,
 //     and may emit multiple event kinds.
 //
+//     For HTTP-backed polling sources, sources.NewHTTPSource provides a shared
+//     helper for generic params:
+//
+//   - params.url
+//
+//   - params.user_agent
+//
+//   - params.conditional (optional, default true)
+//
+//     When conditional polling is enabled, feedkit opportunistically uses ETag
+//     and Last-Modified validators. A 304 Not Modified response is treated as a
+//     successful poll that emits no events.
+//
 //   - scheduler
 //     Runs one goroutine per job:
 //
@@ -64,7 +77,10 @@
 //   - processors
 //     Generic processor interface and named factory registry for wiring chains.
 //
-//   - normalize
+//   - processors/dedupe
+//     Built-in in-memory LRU dedupe processor keyed by Event.ID.
+//
+//   - processors/normalize
 //     Concrete pipeline processor for raw->canonical mapping.
 //     If no normalizer matches, the event passes through unchanged by default.
 //
@@ -73,6 +89,10 @@
 //
 //   - sinks
 //     Sink abstractions + sink registry.
+//     Built-ins include stdout, NATS, and Postgres. For Postgres, downstream
+//     code registers table schemas/mappers while feedkit manages DDL, writes,
+//     optional automatic retention pruning (via sink params.prune), and
+//     manual prune helpers. Postgres table schemas must declare PruneColumn.
 //
 // Typical wiring (daemon main.go)
 //

go.mod

@@ -3,6 +3,7 @@ module gitea.maximumdirect.net/ejr/feedkit
 go 1.22
 
 require (
+	github.com/lib/pq v1.10.9
 	github.com/nats-io/nats.go v1.34.0
 	gopkg.in/yaml.v3 v3.0.1
 )

go.sum

@@ -1,5 +1,7 @@
 github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
 github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
+github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/nats-io/nats.go v1.34.0 h1:fnxnPCNiwIG5w08rlMcEKTUw4AV/nKyGCOJE8TdhSPk=
 github.com/nats-io/nats.go v1.34.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
 github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=

pipeline/dedupe.go

@@ -1,5 +0,0 @@
-package pipeline
-
-// Placeholder for dedupe processor:
-// - key by Event.ID or computed key
-// - in-memory store first; later optional Postgres-backed

processors/dedupe/doc.go

@@ -0,0 +1,28 @@
+// Package dedupe provides a default in-memory LRU deduplication processor.
+//
+// The processor keys strictly by event.Event.ID:
+//   - first-seen IDs pass through
+//   - repeated IDs are dropped
+//
+// The in-memory seen-ID set is bounded by a required maxEntries capacity.
+// When capacity is exceeded, the least recently used ID is evicted.
+//
+// Typical registry wiring:
+//
+// ```go
+// reg := processors.NewRegistry()
+// reg.Register("dedupe", dedupe.Factory(10_000))
+//
+//	reg.Register("normalize", func() (processors.Processor, error) {
+//	    return normalize.NewProcessor(myNormalizers, false), nil
+//	})
+//
+// chain, err := reg.BuildChain([]string{"dedupe", "normalize"})
+//
+//	if err != nil {
+//	    // handle wiring error
+//	}
+//
+// p := &pipeline.Pipeline{Processors: chain}
+// ```
+package dedupe

processors/dedupe/processor.go

@@ -0,0 +1,89 @@
+package dedupe
+
+import (
+	"container/list"
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/feedkit/processors"
+)
+
+// Processor drops duplicate events by Event.ID using an in-memory LRU.
+type Processor struct {
+	maxEntries int
+
+	mu    sync.Mutex
+	order *list.List               // most-recent at front, least-recent at back
+	byID  map[string]*list.Element // id -> list element (element.Value is string id)
+}
+
+var _ processors.Processor = (*Processor)(nil)
+
+// NewProcessor constructs a dedupe processor with a required max entry count.
+func NewProcessor(maxEntries int) (*Processor, error) {
+	if maxEntries <= 0 {
+		return nil, fmt.Errorf("dedupe: maxEntries must be > 0, got %d", maxEntries)
+	}
+
+	return &Processor{
+		maxEntries: maxEntries,
+		order:      list.New(),
+		byID:       make(map[string]*list.Element, maxEntries),
+	}, nil
+}
+
+// Factory returns a processors.Factory that constructs Processor instances.
+func Factory(maxEntries int) processors.Factory {
+	return func() (processors.Processor, error) {
+		return NewProcessor(maxEntries)
+	}
+}
+
+// Process implements processors.Processor.
+func (p *Processor) Process(_ context.Context, in event.Event) (*event.Event, error) {
+	if p == nil {
+		return nil, fmt.Errorf("dedupe: processor is nil")
+	}
+	if p.maxEntries <= 0 {
+		return nil, fmt.Errorf("dedupe: processor maxEntries must be > 0")
+	}
+
+	id := strings.TrimSpace(in.ID)
+	if id == "" {
+		return nil, fmt.Errorf("dedupe: event ID is required")
+	}
+
+	p.mu.Lock()
+
+	if p.order == nil || p.byID == nil {
+		p.mu.Unlock()
+		return nil, fmt.Errorf("dedupe: processor is not initialized")
+	}
+
+	if elem, exists := p.byID[id]; exists {
+		p.order.MoveToFront(elem)
+		p.mu.Unlock()
+		return nil, nil
+	}
+
+	elem := p.order.PushFront(id)
+	p.byID[id] = elem
+
+	if p.order.Len() > p.maxEntries {
+		oldest := p.order.Back()
+		if oldest != nil {
+			p.order.Remove(oldest)
+			if oldestID, ok := oldest.Value.(string); ok {
+				delete(p.byID, oldestID)
+			}
+		}
+	}
+
+	p.mu.Unlock()
+
+	out := in
+	return &out, nil
+}

processors/dedupe/processor_test.go

@@ -0,0 +1,163 @@
+package dedupe
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/feedkit/processors"
+)
+
+func TestNewProcessorValidation(t *testing.T) {
+	t.Run("rejects non-positive maxEntries", func(t *testing.T) {
+		for _, maxEntries := range []int{0, -1} {
+			p, err := NewProcessor(maxEntries)
+			if err == nil {
+				t.Fatalf("expected error for maxEntries=%d, got nil", maxEntries)
+			}
+			if p != nil {
+				t.Fatalf("expected nil processor for maxEntries=%d", maxEntries)
+			}
+			if !strings.Contains(err.Error(), "maxEntries") {
+				t.Fatalf("unexpected error: %v", err)
+			}
+		}
+	})
+
+	t.Run("accepts positive maxEntries", func(t *testing.T) {
+		p, err := NewProcessor(1)
+		if err != nil {
+			t.Fatalf("NewProcessor error: %v", err)
+		}
+		if p == nil {
+			t.Fatalf("expected processor, got nil")
+		}
+	})
+}
+
+func TestProcessorFirstSeenAndDuplicate(t *testing.T) {
+	p, err := NewProcessor(8)
+	if err != nil {
+		t.Fatalf("NewProcessor error: %v", err)
+	}
+
+	ctx := context.Background()
+	first := testEvent("evt-1")
+
+	out, err := p.Process(ctx, first)
+	if err != nil {
+		t.Fatalf("Process first error: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("expected first event to pass through")
+	}
+	if out.ID != first.ID {
+		t.Fatalf("expected unchanged ID %q, got %q", first.ID, out.ID)
+	}
+
+	out, err = p.Process(ctx, first)
+	if err != nil {
+		t.Fatalf("Process duplicate error: %v", err)
+	}
+	if out != nil {
+		t.Fatalf("expected duplicate to be dropped, got %#v", out)
+	}
+
+	out, err = p.Process(ctx, testEvent("evt-2"))
+	if err != nil {
+		t.Fatalf("Process second unique error: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("expected second unique event to pass through")
+	}
+}
+
+func TestProcessorLRUEvictionAndPromotion(t *testing.T) {
+	p, err := NewProcessor(2)
+	if err != nil {
+		t.Fatalf("NewProcessor error: %v", err)
+	}
+
+	ctx := context.Background()
+
+	mustPass(t, p, ctx, "a")
+	mustPass(t, p, ctx, "b")
+	mustDrop(t, p, ctx, "a") // promote "a" so "b" becomes least-recently-used
+	mustPass(t, p, ctx, "c") // evicts "b"
+	mustDrop(t, p, ctx, "a") // "a" should still be tracked after promotion
+	mustPass(t, p, ctx, "b") // "b" was evicted, so now it passes again
+}
+
+func TestProcessorRejectsBlankID(t *testing.T) {
+	p, err := NewProcessor(4)
+	if err != nil {
+		t.Fatalf("NewProcessor error: %v", err)
+	}
+
+	in := testEvent("   ")
+	out, err := p.Process(context.Background(), in)
+	if err == nil {
+		t.Fatalf("expected error for blank ID")
+	}
+	if out != nil {
+		t.Fatalf("expected nil output on error, got %#v", out)
+	}
+	if !strings.Contains(err.Error(), "event ID is required") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestFactoryWithRegistry(t *testing.T) {
+	r := processors.NewRegistry()
+	r.Register("dedupe", Factory(3))
+
+	p, err := r.Build("dedupe")
+	if err != nil {
+		t.Fatalf("Build error: %v", err)
+	}
+	if p == nil {
+		t.Fatalf("expected processor, got nil")
+	}
+
+	out, err := p.Process(context.Background(), testEvent("evt-factory-1"))
+	if err != nil {
+		t.Fatalf("Process error: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("expected first event to pass through")
+	}
+}
+
+func mustPass(t *testing.T, p *Processor, ctx context.Context, id string) {
+	t.Helper()
+	out, err := p.Process(ctx, testEvent(id))
+	if err != nil {
+		t.Fatalf("expected pass for id=%q, got error: %v", id, err)
+	}
+	if out == nil {
+		t.Fatalf("expected pass for id=%q, got drop", id)
+	}
+}
+
+func mustDrop(t *testing.T, p *Processor, ctx context.Context, id string) {
+	t.Helper()
+	out, err := p.Process(ctx, testEvent(id))
+	if err != nil {
+		t.Fatalf("expected drop for id=%q, got error: %v", id, err)
+	}
+	if out != nil {
+		t.Fatalf("expected drop for id=%q, got output", id)
+	}
+}
+
+func testEvent(id string) event.Event {
+	return event.Event{
+		ID:        id,
+		Kind:      event.Kind("observation"),
+		Source:    "source-1",
+		EmittedAt: time.Now().UTC(),
+		Payload:   map[string]any{"ok": true},
+	}
+}

processors/doc.go

@@ -9,11 +9,13 @@
 // Example:
 //
 //	reg := processors.NewRegistry()
+//	reg.Register("dedupe", dedupe.Factory(10_000))
 //	reg.Register("normalize", func() (processors.Processor, error) {
+//	    // import "gitea.maximumdirect.net/ejr/feedkit/processors/normalize"
 //	    return normalize.NewProcessor(myNormalizers, false), nil
 //	})
 //
-//	chain, err := reg.BuildChain([]string{"normalize"})
+//	chain, err := reg.BuildChain([]string{"dedupe", "normalize"})
 //	if err != nil {
 //	    // handle wiring error
 //	}

processors/normalize/helpers.go

@@ -0,0 +1,84 @@
+package normalize
+
+import (
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+// PayloadJSONBytes extracts a JSON payload into bytes suitable for json.Unmarshal.
+//
+// Supported payload shapes:
+//   - json.RawMessage
+//   - []byte
+//   - string
+//   - map[string]any
+func PayloadJSONBytes(e event.Event) ([]byte, error) {
+	if e.Payload == nil {
+		return nil, fmt.Errorf("payload is nil")
+	}
+
+	switch v := e.Payload.(type) {
+	case json.RawMessage:
+		if len(v) == 0 {
+			return nil, fmt.Errorf("payload is empty json.RawMessage")
+		}
+		return []byte(v), nil
+	case []byte:
+		if len(v) == 0 {
+			return nil, fmt.Errorf("payload is empty []byte")
+		}
+		return v, nil
+	case string:
+		if v == "" {
+			return nil, fmt.Errorf("payload is empty string")
+		}
+		return []byte(v), nil
+	case map[string]any:
+		b, err := json.Marshal(v)
+		if err != nil {
+			return nil, fmt.Errorf("marshal map payload: %w", err)
+		}
+		return b, nil
+	default:
+		return nil, fmt.Errorf("unsupported payload type %T", e.Payload)
+	}
+}
+
+// DecodeJSONPayload extracts the event payload as bytes and unmarshals it into T.
+func DecodeJSONPayload[T any](in event.Event) (T, error) {
+	var zero T
+
+	b, err := PayloadJSONBytes(in)
+	if err != nil {
+		return zero, fmt.Errorf("extract payload: %w", err)
+	}
+
+	var parsed T
+	if err := json.Unmarshal(b, &parsed); err != nil {
+		return zero, fmt.Errorf("decode raw payload: %w", err)
+	}
+
+	return parsed, nil
+}
+
+// FinalizeEvent builds the output event envelope by copying the input and applying
+// the new schema/payload, plus optional EffectiveAt.
+func FinalizeEvent(in event.Event, outSchema string, outPayload any, effectiveAt time.Time) (*event.Event, error) {
+	out := in
+	out.Schema = outSchema
+	out.Payload = outPayload
+
+	if !effectiveAt.IsZero() {
+		t := effectiveAt.UTC()
+		out.EffectiveAt = &t
+	}
+
+	if err := out.Validate(); err != nil {
+		return nil, err
+	}
+
+	return &out, nil
+}

processors/normalize/helpers_test.go

@@ -0,0 +1,118 @@
+package normalize
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+func TestPayloadJSONBytesSupportedShapes(t *testing.T) {
+	cases := []struct {
+		name    string
+		payload any
+		want    string
+	}{
+		{name: "rawmessage", payload: json.RawMessage(`{"a":1}`), want: `{"a":1}`},
+		{name: "bytes", payload: []byte(`{"a":2}`), want: `{"a":2}`},
+		{name: "string", payload: `{"a":3}`, want: `{"a":3}`},
+		{name: "map", payload: map[string]any{"a": 4}, want: `{"a":4}`},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got, err := PayloadJSONBytes(event.Event{Payload: tc.payload})
+			if err != nil {
+				t.Fatalf("PayloadJSONBytes() unexpected error: %v", err)
+			}
+			if string(got) != tc.want {
+				t.Fatalf("PayloadJSONBytes() = %s, want %s", string(got), tc.want)
+			}
+		})
+	}
+}
+
+func TestPayloadJSONBytesRejectsInvalidPayloads(t *testing.T) {
+	cases := []struct {
+		name    string
+		payload any
+		want    string
+	}{
+		{name: "nil", payload: nil, want: "payload is nil"},
+		{name: "empty rawmessage", payload: json.RawMessage{}, want: "payload is empty json.RawMessage"},
+		{name: "empty bytes", payload: []byte{}, want: "payload is empty []byte"},
+		{name: "empty string", payload: "", want: "payload is empty string"},
+		{name: "unsupported", payload: 123, want: "unsupported payload type"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := PayloadJSONBytes(event.Event{Payload: tc.payload})
+			if err == nil {
+				t.Fatalf("PayloadJSONBytes() expected error")
+			}
+			if !strings.Contains(err.Error(), tc.want) {
+				t.Fatalf("PayloadJSONBytes() error = %q, want substring %q", err, tc.want)
+			}
+		})
+	}
+}
+
+func TestDecodeJSONPayload(t *testing.T) {
+	type payload struct {
+		Name string `json:"name"`
+	}
+
+	got, err := DecodeJSONPayload[payload](event.Event{
+		Payload: json.RawMessage(`{"name":"alice"}`),
+	})
+	if err != nil {
+		t.Fatalf("DecodeJSONPayload() unexpected error: %v", err)
+	}
+	if got.Name != "alice" {
+		t.Fatalf("DecodeJSONPayload() = %#v, want name alice", got)
+	}
+}
+
+func TestFinalizeEventPreservesEnvelopeAndEffectiveAtBehavior(t *testing.T) {
+	existingEffectiveAt := time.Date(2026, 3, 28, 11, 0, 0, 0, time.UTC)
+	in := event.Event{
+		ID:          "evt-1",
+		Kind:        event.Kind("observation"),
+		Source:      "source-a",
+		EmittedAt:   time.Date(2026, 3, 28, 12, 0, 0, 0, time.UTC),
+		EffectiveAt: &existingEffectiveAt,
+		Schema:      "raw.example.v1",
+		Payload:     map[string]any{"old": true},
+	}
+
+	out, err := FinalizeEvent(in, "example.v1", map[string]any{"value": 1.234567}, time.Time{})
+	if err != nil {
+		t.Fatalf("FinalizeEvent() unexpected error: %v", err)
+	}
+	if out.ID != in.ID || out.Kind != in.Kind || out.Source != in.Source || out.EmittedAt != in.EmittedAt {
+		t.Fatalf("FinalizeEvent() changed preserved envelope fields: %#v", out)
+	}
+	if out.EffectiveAt == nil || !out.EffectiveAt.Equal(existingEffectiveAt) {
+		t.Fatalf("FinalizeEvent() effectiveAt = %#v, want preserved existing value", out.EffectiveAt)
+	}
+
+	nextEffectiveAt := time.Date(2026, 3, 28, 13, 0, 0, 0, time.FixedZone("x", -4*3600))
+	out, err = FinalizeEvent(in, "example.v1", map[string]any{"value": 1.234567}, nextEffectiveAt)
+	if err != nil {
+		t.Fatalf("FinalizeEvent() unexpected overwrite error: %v", err)
+	}
+	if out.EffectiveAt == nil || !out.EffectiveAt.Equal(nextEffectiveAt.UTC()) {
+		t.Fatalf("FinalizeEvent() effectiveAt = %#v, want %s", out.EffectiveAt, nextEffectiveAt.UTC())
+	}
+
+	payloadMap, ok := out.Payload.(map[string]any)
+	if !ok {
+		t.Fatalf("FinalizeEvent() payload type = %T, want map[string]any", out.Payload)
+	}
+	if payloadMap["value"] != 1.234567 {
+		t.Fatalf("FinalizeEvent() payload value = %#v, want unrounded 1.234567", payloadMap["value"])
+	}
+}

sinks/doc.go

@@ -0,0 +1,90 @@
+// Package sinks provides sink abstractions, a sink driver registry, and several
+// built-in sink drivers.
+//
+// Built-in drivers:
+//   - stdout
+//   - nats
+//   - postgres
+//
+// # NATS built-in overview
+//
+// The NATS sink publishes each event as JSON to a configured subject.
+//
+// Required params:
+//   - url: NATS server URL (for example, nats://localhost:4222)
+//   - exchange: NATS subject to publish to
+//
+// Example config:
+//
+//	sinks:
+//	  - name: nats_main
+//	    driver: nats
+//	    params:
+//	      url: nats://localhost:4222
+//	      exchange: feedkit.events
+//
+// # Postgres built-in overview
+//
+// The postgres sink is intentionally split between downstream daemon ownership
+// and feedkit ownership:
+//   - downstream code registers table schema + event mapping functions
+//   - feedkit manages DB connection, create-if-missing DDL, transactional
+//     inserts, optional automatic retention pruning, and manual prune helpers
+//
+// Example config:
+//
+//	sinks:
+//	  - name: pg_main
+//	    driver: postgres
+//	    params:
+//	      uri: postgres://localhost:5432/feedkit?sslmode=disable
+//	      username: feedkit_user
+//	      password: feedkit_pass
+//	      prune: 3d # optional: prune rows older than now-3d on each write tx
+//
+// params.prune supports:
+//   - Go duration strings (72h, 90m, 30s, ...)
+//   - day/week suffixes (3d, 2w)
+//
+// If params.prune is omitted, automatic pruning is disabled.
+//
+// Example downstream wiring:
+//
+//	sinks.MustRegisterPostgresSchema("pg_main", sinks.PostgresSchema{
+//	    Tables: []sinks.PostgresTable{
+//	        {
+//	            Name: "events",
+//	            Columns: []sinks.PostgresColumn{
+//	                {Name: "event_id", Type: "TEXT", Nullable: false},
+//	                {Name: "emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
+//	                {Name: "payload_json", Type: "JSONB", Nullable: false},
+//	            },
+//	            PrimaryKey:  []string{"event_id"},
+//	            PruneColumn: "emitted_at", // required for retention pruning
+//	        },
+//	    },
+//	    MapEvent: func(ctx context.Context, e event.Event) ([]sinks.PostgresWrite, error) {
+//	        b, err := json.Marshal(e.Payload)
+//	        if err != nil {
+//	            return nil, err
+//	        }
+//	        return []sinks.PostgresWrite{
+//	            {
+//	                Table: "events",
+//	                Values: map[string]any{
+//	                    "event_id":     e.ID,
+//	                    "emitted_at":   e.EmittedAt,
+//	                    "payload_json": string(b),
+//	                },
+//	            },
+//	        }, nil
+//	    },
+//	})
+//
+// Manual pruning via type assertion (administrative helpers):
+//
+//	if p, ok := sink.(sinks.PostgresPruner); ok {
+//	    _, _ = p.PruneKeepLatest(ctx, "events", 10000)
+//	    _, _ = p.PruneOlderThan(ctx, "events", time.Now().UTC().AddDate(0, -1, 0))
+//	}
+package sinks

sinks/helpers.go

@@ -0,0 +1,27 @@
+package sinks
+
+import (
+	"fmt"
+	"strings"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+)
+
+// RegisterPostgresSchemaForConfiguredSinks registers one Postgres schema for each
+// configured sink using driver=postgres.
+func RegisterPostgresSchemaForConfiguredSinks(cfg *config.Config, schema PostgresSchema) error {
+	if cfg == nil {
+		return fmt.Errorf("register postgres schemas: config is nil")
+	}
+
+	for i, sk := range cfg.Sinks {
+		if !strings.EqualFold(strings.TrimSpace(sk.Driver), "postgres") {
+			continue
+		}
+		if err := RegisterPostgresSchema(sk.Name, schema); err != nil {
+			return fmt.Errorf("register postgres schema for sinks[%d] name=%q: %w", i, sk.Name, err)
+		}
+	}
+
+	return nil
+}

sinks/helpers_test.go

@@ -0,0 +1,86 @@
+package sinks
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+func TestRegisterPostgresSchemaForConfiguredSinksNilConfig(t *testing.T) {
+	err := RegisterPostgresSchemaForConfiguredSinks(nil, testPostgresSchema())
+	if err == nil {
+		t.Fatalf("RegisterPostgresSchemaForConfiguredSinks(nil) expected error")
+	}
+	if !strings.Contains(err.Error(), "config is nil") {
+		t.Fatalf("error = %q, want config is nil", err)
+	}
+}
+
+func TestRegisterPostgresSchemaForConfiguredSinksNonPostgresNoOp(t *testing.T) {
+	cfg := &config.Config{
+		Sinks: []config.SinkConfig{
+			{Name: uniqueSinkName("stdout"), Driver: "stdout"},
+			{Name: uniqueSinkName("nats"), Driver: "nats"},
+		},
+	}
+
+	if err := RegisterPostgresSchemaForConfiguredSinks(cfg, testPostgresSchema()); err != nil {
+		t.Fatalf("RegisterPostgresSchemaForConfiguredSinks(non-postgres) error = %v", err)
+	}
+}
+
+func TestRegisterPostgresSchemaForConfiguredSinksDuplicateRegistrationFails(t *testing.T) {
+	cfg := &config.Config{
+		Sinks: []config.SinkConfig{
+			{Name: uniqueSinkName("pg"), Driver: "postgres"},
+		},
+	}
+
+	if err := RegisterPostgresSchemaForConfiguredSinks(cfg, testPostgresSchema()); err != nil {
+		t.Fatalf("first RegisterPostgresSchemaForConfiguredSinks() error = %v", err)
+	}
+
+	err := RegisterPostgresSchemaForConfiguredSinks(cfg, testPostgresSchema())
+	if err == nil {
+		t.Fatalf("second RegisterPostgresSchemaForConfiguredSinks() expected duplicate error")
+	}
+	if !strings.Contains(err.Error(), "already registered") {
+		t.Fatalf("error = %q, want already registered", err)
+	}
+}
+
+func testPostgresSchema() PostgresSchema {
+	return PostgresSchema{
+		Tables: []PostgresTable{
+			{
+				Name: "events",
+				Columns: []PostgresColumn{
+					{Name: "event_id", Type: "TEXT", Nullable: false},
+					{Name: "emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
+				},
+				PrimaryKey:  []string{"event_id"},
+				PruneColumn: "emitted_at",
+			},
+		},
+		MapEvent: func(_ context.Context, e event.Event) ([]PostgresWrite, error) {
+			return []PostgresWrite{
+				{
+					Table: "events",
+					Values: map[string]any{
+						"event_id":   e.ID,
+						"emitted_at": e.EmittedAt,
+					},
+				},
+			}, nil
+		},
+	}
+}
+
+func uniqueSinkName(prefix string) string {
+	return fmt.Sprintf("%s_%d", prefix, time.Now().UnixNano())
+}

sinks/postgres.go

@@ -2,36 +2,460 @@ package sinks
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
 
 	"gitea.maximumdirect.net/ejr/feedkit/config"
 	"gitea.maximumdirect.net/ejr/feedkit/event"
+	_ "github.com/lib/pq"
 )
 
-type PostgresSink struct {
+const postgresInitTimeout = 5 * time.Second
-	name string
+
-	dsn  string
+type postgresTx interface {
+	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+	Commit() error
+	Rollback() error
 }
 
-func NewPostgresSinkFromConfig(cfg config.SinkConfig) (Sink, error) {
+type postgresExecer interface {
-	dsn, err := requireStringParam(cfg, "dsn")
+	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+}
+
+type postgresDB interface {
+	PingContext(ctx context.Context) error
+	BeginTx(ctx context.Context, opts *sql.TxOptions) (postgresTx, error)
+	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
+	Close() error
+}
+
+type sqlDBWrapper struct {
+	db *sql.DB
+}
+
+func (w *sqlDBWrapper) PingContext(ctx context.Context) error {
+	return w.db.PingContext(ctx)
+}
+
+func (w *sqlDBWrapper) BeginTx(ctx context.Context, opts *sql.TxOptions) (postgresTx, error) {
+	tx, err := w.db.BeginTx(ctx, opts)
 	if err != nil {
 		return nil, err
 	}
-	return &PostgresSink{name: cfg.Name, dsn: dsn}, nil
+	return &sqlTxWrapper{tx: tx}, nil
+}
+
+func (w *sqlDBWrapper) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
+	return w.db.ExecContext(ctx, query, args...)
+}
+
+func (w *sqlDBWrapper) Close() error {
+	return w.db.Close()
+}
+
+type sqlTxWrapper struct {
+	tx *sql.Tx
+}
+
+func (w *sqlTxWrapper) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
+	return w.tx.ExecContext(ctx, query, args...)
+}
+
+func (w *sqlTxWrapper) Commit() error {
+	return w.tx.Commit()
+}
+
+func (w *sqlTxWrapper) Rollback() error {
+	return w.tx.Rollback()
+}
+
+var openPostgresDB = func(dsn string) (postgresDB, error) {
+	db, err := sql.Open("postgres", dsn)
+	if err != nil {
+		return nil, err
+	}
+	return &sqlDBWrapper{db: db}, nil
+}
+
+type PostgresSink struct {
+	name        string
+	db          postgresDB
+	schema      postgresSchemaCompiled
+	pruneWindow time.Duration
+}
+
+func NewPostgresSinkFromConfig(cfg config.SinkConfig) (Sink, error) {
+	uri, err := requireStringParam(cfg, "uri")
+	if err != nil {
+		return nil, err
+	}
+	username, err := requireStringParam(cfg, "username")
+	if err != nil {
+		return nil, err
+	}
+	password, err := requireStringParam(cfg, "password")
+	if err != nil {
+		return nil, err
+	}
+	pruneWindow, err := parsePostgresPruneWindow(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	schema, ok := lookupPostgresSchema(cfg.Name)
+	if !ok {
+		return nil, fmt.Errorf("postgres sink %q: no schema registered (call sinks.RegisterPostgresSchema before building sinks)", cfg.Name)
+	}
+
+	dsn, err := buildPostgresDSN(uri, username, password)
+	if err != nil {
+		return nil, fmt.Errorf("postgres sink %q: build dsn: %w", cfg.Name, err)
+	}
+
+	db, err := openPostgresDB(dsn)
+	if err != nil {
+		return nil, fmt.Errorf("postgres sink %q: open db: %w", cfg.Name, err)
+	}
+
+	s := &PostgresSink{name: cfg.Name, db: db, schema: schema, pruneWindow: pruneWindow}
+	if err := s.initialize(); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+
+	return s, nil
 }
 
 func (p *PostgresSink) Name() string { return p.name }
 
 func (p *PostgresSink) Consume(ctx context.Context, e event.Event) error {
-	_ = ctx
-
 	// Boundary validation: if something upstream violated invariants,
-	// surface it loudly rather than printing partial nonsense.
+	// surface it loudly rather than writing corrupt rows.
 	if err := e.Validate(); err != nil {
 		return fmt.Errorf("postgres sink: invalid event: %w", err)
 	}
 
-	// TODO implement Postgres transaction
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
+	writes, err := p.schema.mapEvent(ctx, e)
+	if err != nil {
+		return fmt.Errorf("postgres sink: map event: %w", err)
+	}
+	if len(writes) == 0 {
+		return nil
+	}
+
+	tx, err := p.db.BeginTx(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("postgres sink: begin tx: %w", err)
+	}
+
+	committed := false
+	defer func() {
+		if !committed {
+			_ = tx.Rollback()
+		}
+	}()
+
+	for _, w := range writes {
+		tbl, err := p.schema.validateWrite(w)
+		if err != nil {
+			return fmt.Errorf("postgres sink: %w", err)
+		}
+
+		query, args, err := buildInsertSQL(tbl, w)
+		if err != nil {
+			return fmt.Errorf("postgres sink: build insert for table %q: %w", tbl.name, err)
+		}
+		if _, err := tx.ExecContext(ctx, query, args...); err != nil {
+			return fmt.Errorf("postgres sink: insert into %q: %w", tbl.name, err)
+		}
+	}
+	if p.pruneWindow > 0 {
+		cutoff := time.Now().UTC().Add(-p.pruneWindow)
+		for _, tableName := range p.schema.tableOrder {
+			tbl := p.schema.tables[tableName]
+			if _, err := execPruneOlderThan(ctx, tx, tbl, cutoff); err != nil {
+				return fmt.Errorf("postgres sink: %w", err)
+			}
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("postgres sink: commit tx: %w", err)
+	}
+	committed = true
+
 	return nil
 }
+
+func (p *PostgresSink) PruneKeepLatest(ctx context.Context, table string, keep int) (int64, error) {
+	if keep < 0 {
+		return 0, fmt.Errorf("postgres sink: keep must be >= 0")
+	}
+	tbl, err := p.lookupTable(table)
+	if err != nil {
+		return 0, err
+	}
+
+	query := fmt.Sprintf(
+		`DELETE FROM %s WHERE ctid IN (
+	SELECT ctid FROM %s
+	ORDER BY %s DESC
+	OFFSET $1
+)`,
+		quotePostgresIdent(tbl.name),
+		quotePostgresIdent(tbl.name),
+		quotePostgresIdent(tbl.pruneColumn),
+	)
+
+	res, err := p.db.ExecContext(ctx, query, keep)
+	if err != nil {
+		return 0, fmt.Errorf("postgres sink: prune keep latest table %q: %w", tbl.name, err)
+	}
+	rows, err := res.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("postgres sink: prune keep latest table %q rows affected: %w", tbl.name, err)
+	}
+	return rows, nil
+}
+
+func (p *PostgresSink) PruneOlderThan(ctx context.Context, table string, cutoff time.Time) (int64, error) {
+	tbl, err := p.lookupTable(table)
+	if err != nil {
+		return 0, err
+	}
+
+	rows, err := execPruneOlderThan(ctx, p.db, tbl, cutoff)
+	if err != nil {
+		return 0, fmt.Errorf("postgres sink: %w", err)
+	}
+	return rows, nil
+}
+
+func (p *PostgresSink) PruneAllKeepLatest(ctx context.Context, keep int) (map[string]int64, error) {
+	counts := make(map[string]int64, len(p.schema.tableOrder))
+	for _, table := range p.schema.tableOrder {
+		n, err := p.PruneKeepLatest(ctx, table, keep)
+		if err != nil {
+			return counts, err
+		}
+		counts[table] = n
+	}
+	return counts, nil
+}
+
+func (p *PostgresSink) PruneAllOlderThan(ctx context.Context, cutoff time.Time) (map[string]int64, error) {
+	counts := make(map[string]int64, len(p.schema.tableOrder))
+	for _, table := range p.schema.tableOrder {
+		n, err := p.PruneOlderThan(ctx, table, cutoff)
+		if err != nil {
+			return counts, err
+		}
+		counts[table] = n
+	}
+	return counts, nil
+}
+
+func (p *PostgresSink) initialize() error {
+	ctx, cancel := context.WithTimeout(context.Background(), postgresInitTimeout)
+	defer cancel()
+
+	if err := p.db.PingContext(ctx); err != nil {
+		return fmt.Errorf("postgres sink %q: ping db: %w", p.name, err)
+	}
+
+	for _, tableName := range p.schema.tableOrder {
+		tbl := p.schema.tables[tableName]
+
+		createTableSQL := buildCreateTableSQL(tbl)
+		if _, err := p.db.ExecContext(ctx, createTableSQL); err != nil {
+			return fmt.Errorf("postgres sink %q: ensure table %q: %w", p.name, tbl.name, err)
+		}
+
+		for _, idx := range tbl.indexes {
+			createIndexSQL := buildCreateIndexSQL(tbl.name, idx)
+			if _, err := p.db.ExecContext(ctx, createIndexSQL); err != nil {
+				return fmt.Errorf("postgres sink %q: ensure index %q on %q: %w", p.name, idx.Name, tbl.name, err)
+			}
+		}
+	}
+
+	return nil
+}
+
+func (p *PostgresSink) lookupTable(table string) (postgresTableCompiled, error) {
+	table = strings.TrimSpace(table)
+	if table == "" {
+		return postgresTableCompiled{}, fmt.Errorf("postgres sink: table cannot be empty")
+	}
+	tbl, ok := p.schema.tables[table]
+	if !ok {
+		return postgresTableCompiled{}, fmt.Errorf("postgres sink: unknown table %q", table)
+	}
+	return tbl, nil
+}
+
+func buildPostgresDSN(uri, username, password string) (string, error) {
+	u, err := url.Parse(strings.TrimSpace(uri))
+	if err != nil {
+		return "", fmt.Errorf("invalid uri: %w", err)
+	}
+	if u.Scheme == "" {
+		return "", fmt.Errorf("invalid uri: missing scheme")
+	}
+	if u.Host == "" {
+		return "", fmt.Errorf("invalid uri: missing host")
+	}
+	u.User = url.UserPassword(username, password)
+	return u.String(), nil
+}
+
+func parsePostgresPruneWindow(cfg config.SinkConfig) (time.Duration, error) {
+	raw, ok := cfg.Params["prune"]
+	if !ok || raw == nil {
+		return 0, nil
+	}
+
+	s, ok := raw.(string)
+	if !ok {
+		return 0, fmt.Errorf("sink %q: params.prune must be a string duration (e.g. 72h, 3d, 2w)", cfg.Name)
+	}
+
+	d, err := parsePostgresPruneDuration(s)
+	if err != nil {
+		return 0, fmt.Errorf("sink %q: params.prune %q is invalid: %w", cfg.Name, s, err)
+	}
+	return d, nil
+}
+
+func parsePostgresPruneDuration(raw string) (time.Duration, error) {
+	s := strings.TrimSpace(raw)
+	if s == "" {
+		return 0, fmt.Errorf("must not be empty")
+	}
+
+	lower := strings.ToLower(s)
+	if strings.HasSuffix(lower, "d") || strings.HasSuffix(lower, "w") {
+		unit := lower[len(lower)-1]
+		n, err := strconv.Atoi(strings.TrimSpace(lower[:len(lower)-1]))
+		if err != nil {
+			return 0, fmt.Errorf("must use a positive integer before %q", string(unit))
+		}
+		if n <= 0 {
+			return 0, fmt.Errorf("must be > 0")
+		}
+		if unit == 'd' {
+			return time.Duration(n) * 24 * time.Hour, nil
+		}
+		return time.Duration(n) * 7 * 24 * time.Hour, nil
+	}
+
+	d, err := time.ParseDuration(s)
+	if err != nil {
+		return 0, fmt.Errorf("must be a Go duration or use d/w suffixes")
+	}
+	if d <= 0 {
+		return 0, fmt.Errorf("must be > 0")
+	}
+	return d, nil
+}
+
+func buildPruneOlderThanSQL(tbl postgresTableCompiled) string {
+	return fmt.Sprintf(
+		`DELETE FROM %s WHERE %s < $1`,
+		quotePostgresIdent(tbl.name),
+		quotePostgresIdent(tbl.pruneColumn),
+	)
+}
+
+func execPruneOlderThan(ctx context.Context, execer postgresExecer, tbl postgresTableCompiled, cutoff time.Time) (int64, error) {
+	query := buildPruneOlderThanSQL(tbl)
+	res, err := execer.ExecContext(ctx, query, cutoff)
+	if err != nil {
+		return 0, fmt.Errorf("prune older than table %q: %w", tbl.name, err)
+	}
+	rows, err := res.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("prune older than table %q rows affected: %w", tbl.name, err)
+	}
+	return rows, nil
+}
+
+func buildCreateTableSQL(tbl postgresTableCompiled) string {
+	defs := make([]string, 0, len(tbl.columnOrder)+1)
+	for _, colName := range tbl.columnOrder {
+		col := tbl.columns[colName]
+		def := fmt.Sprintf("%s %s", quotePostgresIdent(col.Name), col.Type)
+		if !col.Nullable {
+			def += " NOT NULL"
+		}
+		defs = append(defs, def)
+	}
+	if len(tbl.primaryKey) > 0 {
+		defs = append(defs, fmt.Sprintf("PRIMARY KEY (%s)", joinQuotedIdents(tbl.primaryKey)))
+	}
+
+	return fmt.Sprintf(
+		"CREATE TABLE IF NOT EXISTS %s (%s)",
+		quotePostgresIdent(tbl.name),
+		strings.Join(defs, ", "),
+	)
+}
+
+func buildCreateIndexSQL(tableName string, idx PostgresIndex) string {
+	unique := ""
+	if idx.Unique {
+		unique = "UNIQUE "
+	}
+
+	return fmt.Sprintf(
+		"CREATE %sINDEX IF NOT EXISTS %s ON %s (%s)",
+		unique,
+		quotePostgresIdent(idx.Name),
+		quotePostgresIdent(tableName),
+		joinQuotedIdents(idx.Columns),
+	)
+}
+
+func buildInsertSQL(tbl postgresTableCompiled, w PostgresWrite) (string, []any, error) {
+	cols := make([]string, 0, len(tbl.columnOrder))
+	args := make([]any, 0, len(tbl.columnOrder))
+	placeholders := make([]string, 0, len(tbl.columnOrder))
+
+	for i, colName := range tbl.columnOrder {
+		v, ok := w.Values[colName]
+		if !ok {
+			return "", nil, fmt.Errorf("missing value for column %q", colName)
+		}
+		cols = append(cols, quotePostgresIdent(colName))
+		args = append(args, v)
+		placeholders = append(placeholders, "$"+strconv.Itoa(i+1))
+	}
+
+	q := fmt.Sprintf(
+		"INSERT INTO %s (%s) VALUES (%s)",
+		quotePostgresIdent(tbl.name),
+		strings.Join(cols, ", "),
+		strings.Join(placeholders, ", "),
+	)
+	return q, args, nil
+}
+
+func joinQuotedIdents(idents []string) string {
+	quoted := make([]string, 0, len(idents))
+	for _, s := range idents {
+		quoted = append(quoted, quotePostgresIdent(s))
+	}
+	return strings.Join(quoted, ", ")
+}
+
+func quotePostgresIdent(s string) string {
+	return `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
+}

sinks/postgres_schema.go

@@ -0,0 +1,285 @@
+package sinks
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+// PostgresMapFunc maps one event into zero or more table writes.
+//
+// Returning zero writes means "this event is not mapped for this sink" and is
+// treated as a non-error no-op.
+type PostgresMapFunc func(ctx context.Context, e event.Event) ([]PostgresWrite, error)
+
+// PostgresSchema describes the downstream-provided relational model and mapper
+// for one configured postgres sink.
+type PostgresSchema struct {
+	Tables   []PostgresTable
+	MapEvent PostgresMapFunc
+}
+
+type PostgresWrite struct {
+	Table  string
+	Values map[string]any
+}
+
+type PostgresTable struct {
+	Name        string
+	Columns     []PostgresColumn
+	PrimaryKey  []string
+	PruneColumn string
+	Indexes     []PostgresIndex
+}
+
+type PostgresColumn struct {
+	Name     string
+	Type     string
+	Nullable bool
+}
+
+type PostgresIndex struct {
+	Name    string
+	Columns []string
+	Unique  bool
+}
+
+// PostgresPruner is an optional interface exposed by PostgresSink so downstream
+// applications can call retention helpers via type assertion.
+type PostgresPruner interface {
+	PruneKeepLatest(ctx context.Context, table string, keep int) (int64, error)
+	PruneOlderThan(ctx context.Context, table string, cutoff time.Time) (int64, error)
+	PruneAllKeepLatest(ctx context.Context, keep int) (map[string]int64, error)
+	PruneAllOlderThan(ctx context.Context, cutoff time.Time) (map[string]int64, error)
+}
+
+type postgresSchemaCompiled struct {
+	tableOrder []string
+	tables     map[string]postgresTableCompiled
+	mapEvent   PostgresMapFunc
+}
+
+type postgresTableCompiled struct {
+	name        string
+	columns     map[string]PostgresColumn
+	columnOrder []string
+	primaryKey  []string
+	pruneColumn string
+	indexes     []PostgresIndex
+}
+
+var (
+	postgresSchemaRegistryMu sync.RWMutex
+	postgresSchemaRegistry   = map[string]postgresSchemaCompiled{}
+)
+
+// RegisterPostgresSchema registers one downstream schema by sink name.
+//
+// This should be called by downstream daemon wiring code before sink
+// construction. Duplicate sink-name registrations are rejected.
+func RegisterPostgresSchema(sinkName string, schema PostgresSchema) error {
+	sinkName = strings.TrimSpace(sinkName)
+	if sinkName == "" {
+		return fmt.Errorf("postgres schema: sink name cannot be empty")
+	}
+
+	compiled, err := compilePostgresSchema(schema)
+	if err != nil {
+		return err
+	}
+
+	postgresSchemaRegistryMu.Lock()
+	defer postgresSchemaRegistryMu.Unlock()
+
+	if _, exists := postgresSchemaRegistry[sinkName]; exists {
+		return fmt.Errorf("postgres schema: sink %q already registered", sinkName)
+	}
+
+	postgresSchemaRegistry[sinkName] = compiled
+	return nil
+}
+
+func MustRegisterPostgresSchema(sinkName string, schema PostgresSchema) {
+	if err := RegisterPostgresSchema(sinkName, schema); err != nil {
+		panic(err)
+	}
+}
+
+func lookupPostgresSchema(sinkName string) (postgresSchemaCompiled, bool) {
+	postgresSchemaRegistryMu.RLock()
+	defer postgresSchemaRegistryMu.RUnlock()
+
+	s, ok := postgresSchemaRegistry[sinkName]
+	return s, ok
+}
+
+func compilePostgresSchema(schema PostgresSchema) (postgresSchemaCompiled, error) {
+	if schema.MapEvent == nil {
+		return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: map function is required")
+	}
+	if len(schema.Tables) == 0 {
+		return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: at least one table is required")
+	}
+
+	compiled := postgresSchemaCompiled{
+		tableOrder: make([]string, 0, len(schema.Tables)),
+		tables:     make(map[string]postgresTableCompiled, len(schema.Tables)),
+		mapEvent:   schema.MapEvent,
+	}
+
+	seenTables := map[string]bool{}
+	seenIndexes := map[string]bool{}
+
+	for i, tbl := range schema.Tables {
+		tableName := strings.TrimSpace(tbl.Name)
+		if tableName == "" {
+			return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: tables[%d].name is required", i)
+		}
+		if seenTables[tableName] {
+			return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: duplicate table name %q", tableName)
+		}
+		seenTables[tableName] = true
+
+		if len(tbl.Columns) == 0 {
+			return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: table %q must define at least one column", tableName)
+		}
+
+		colOrder := make([]string, 0, len(tbl.Columns))
+		colMap := make(map[string]PostgresColumn, len(tbl.Columns))
+		for j, col := range tbl.Columns {
+			colName := strings.TrimSpace(col.Name)
+			if colName == "" {
+				return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: table %q columns[%d].name is required", tableName, j)
+			}
+			if _, exists := colMap[colName]; exists {
+				return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: table %q duplicate column %q", tableName, colName)
+			}
+			if strings.TrimSpace(col.Type) == "" {
+				return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: table %q column %q type is required", tableName, colName)
+			}
+			colOrder = append(colOrder, colName)
+			colMap[colName] = PostgresColumn{
+				Name:     colName,
+				Type:     strings.TrimSpace(col.Type),
+				Nullable: col.Nullable,
+			}
+		}
+
+		pk, err := validatePostgresColumnSet(tableName, "primary key", tbl.PrimaryKey, colMap)
+		if err != nil {
+			return postgresSchemaCompiled{}, err
+		}
+
+		pruneCol := strings.TrimSpace(tbl.PruneColumn)
+		if pruneCol == "" {
+			return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: table %q prune column is required", tableName)
+		}
+		if _, ok := colMap[pruneCol]; !ok {
+			return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: table %q prune column %q not found in columns", tableName, pruneCol)
+		}
+
+		indexes := make([]PostgresIndex, 0, len(tbl.Indexes))
+		for j, idx := range tbl.Indexes {
+			idxName := strings.TrimSpace(idx.Name)
+			if idxName == "" {
+				return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: table %q indexes[%d].name is required", tableName, j)
+			}
+			if len(idx.Columns) == 0 {
+				return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: table %q index %q must include at least one column", tableName, idxName)
+			}
+			if seenIndexes[idxName] {
+				return postgresSchemaCompiled{}, fmt.Errorf("postgres schema: duplicate index name %q", idxName)
+			}
+			seenIndexes[idxName] = true
+
+			idxCols, err := validatePostgresColumnSet(tableName, fmt.Sprintf("index %q columns", idxName), idx.Columns, colMap)
+			if err != nil {
+				return postgresSchemaCompiled{}, err
+			}
+
+			indexes = append(indexes, PostgresIndex{
+				Name:    idxName,
+				Columns: idxCols,
+				Unique:  idx.Unique,
+			})
+		}
+
+		compiled.tableOrder = append(compiled.tableOrder, tableName)
+		compiled.tables[tableName] = postgresTableCompiled{
+			name:        tableName,
+			columns:     colMap,
+			columnOrder: colOrder,
+			primaryKey:  pk,
+			pruneColumn: pruneCol,
+			indexes:     indexes,
+		}
+	}
+
+	return compiled, nil
+}
+
+func validatePostgresColumnSet(tableName, label string, cols []string, colMap map[string]PostgresColumn) ([]string, error) {
+	if len(cols) == 0 {
+		return nil, nil
+	}
+	out := make([]string, 0, len(cols))
+	seen := map[string]bool{}
+	for _, c := range cols {
+		name := strings.TrimSpace(c)
+		if name == "" {
+			return nil, fmt.Errorf("postgres schema: table %q %s contains empty column name", tableName, label)
+		}
+		if seen[name] {
+			return nil, fmt.Errorf("postgres schema: table %q %s contains duplicate column %q", tableName, label, name)
+		}
+		if _, ok := colMap[name]; !ok {
+			return nil, fmt.Errorf("postgres schema: table %q %s references unknown column %q", tableName, label, name)
+		}
+		seen[name] = true
+		out = append(out, name)
+	}
+	return out, nil
+}
+
+func (s postgresSchemaCompiled) validateWrite(w PostgresWrite) (postgresTableCompiled, error) {
+	tableName := strings.TrimSpace(w.Table)
+	if tableName == "" {
+		return postgresTableCompiled{}, fmt.Errorf("write table is required")
+	}
+	t, ok := s.tables[tableName]
+	if !ok {
+		return postgresTableCompiled{}, fmt.Errorf("table %q is not defined in postgres schema", tableName)
+	}
+
+	if len(w.Values) == 0 {
+		return postgresTableCompiled{}, fmt.Errorf("write for table %q must include values", tableName)
+	}
+
+	for k := range w.Values {
+		if _, ok := t.columns[k]; !ok {
+			return postgresTableCompiled{}, fmt.Errorf("write for table %q includes unknown column %q", tableName, k)
+		}
+	}
+
+	if len(w.Values) != len(t.columnOrder) {
+		return postgresTableCompiled{}, fmt.Errorf("write for table %q must include all declared columns", tableName)
+	}
+
+	for _, col := range t.columnOrder {
+		v, ok := w.Values[col]
+		if !ok {
+			return postgresTableCompiled{}, fmt.Errorf("write for table %q is missing column %q", tableName, col)
+		}
+		if v == nil {
+			if c := t.columns[col]; !c.Nullable {
+				return postgresTableCompiled{}, fmt.Errorf("write for table %q has nil value for non-null column %q", tableName, col)
+			}
+		}
+	}
+
+	return t, nil
+}

sinks/postgres_test.go

@@ -0,0 +1,743 @@
+package sinks
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+type fakeResult struct {
+	rows int64
+}
+
+func (r fakeResult) LastInsertId() (int64, error) { return 0, errors.New("unsupported") }
+func (r fakeResult) RowsAffected() (int64, error) { return r.rows, nil }
+
+type execCall struct {
+	query string
+	args  []any
+}
+
+type fakeTx struct {
+	execCalls     []execCall
+	execErr       error
+	execErrOnCall int
+	execRows      int64
+	commitErr     error
+	rollbackErr   error
+	commitCalls   int
+	rollbackCalls int
+}
+
+func (t *fakeTx) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) {
+	t.execCalls = append(t.execCalls, execCall{query: query, args: append([]any(nil), args...)})
+	if t.execErr != nil && t.execErrOnCall == len(t.execCalls) {
+		return nil, t.execErr
+	}
+	return fakeResult{rows: t.execRows}, nil
+}
+
+func (t *fakeTx) Commit() error {
+	t.commitCalls++
+	return t.commitErr
+}
+
+func (t *fakeTx) Rollback() error {
+	t.rollbackCalls++
+	return t.rollbackErr
+}
+
+type fakeDB struct {
+	pingErr       error
+	beginErr      error
+	execErr       error
+	execErrOnCall int
+	execRows      int64
+	pingCalls     int
+	beginCalls    int
+	execCalls     []execCall
+	closeCalls    int
+	tx            *fakeTx
+}
+
+func (d *fakeDB) PingContext(_ context.Context) error {
+	d.pingCalls++
+	return d.pingErr
+}
+
+func (d *fakeDB) BeginTx(_ context.Context, _ *sql.TxOptions) (postgresTx, error) {
+	d.beginCalls++
+	if d.beginErr != nil {
+		return nil, d.beginErr
+	}
+	if d.tx == nil {
+		d.tx = &fakeTx{}
+	}
+	return d.tx, nil
+}
+
+func (d *fakeDB) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) {
+	d.execCalls = append(d.execCalls, execCall{query: query, args: append([]any(nil), args...)})
+	if d.execErr != nil && d.execErrOnCall == len(d.execCalls) {
+		return nil, d.execErr
+	}
+	return fakeResult{rows: d.execRows}, nil
+}
+
+func (d *fakeDB) Close() error {
+	d.closeCalls++
+	return nil
+}
+
+func resetPostgresSchemaRegistryForTest() {
+	postgresSchemaRegistryMu.Lock()
+	defer postgresSchemaRegistryMu.Unlock()
+	postgresSchemaRegistry = map[string]postgresSchemaCompiled{}
+}
+
+func withPostgresTestState(t *testing.T) {
+	t.Helper()
+
+	resetPostgresSchemaRegistryForTest()
+	oldOpen := openPostgresDB
+	t.Cleanup(func() {
+		openPostgresDB = oldOpen
+		resetPostgresSchemaRegistryForTest()
+	})
+}
+
+func validTestEvent() event.Event {
+	now := time.Now().UTC()
+	return event.Event{
+		ID:        "evt-1",
+		Kind:      event.Kind("observation"),
+		Source:    "source-1",
+		EmittedAt: now,
+		Payload: map[string]any{
+			"x": 1,
+		},
+	}
+}
+
+func schemaOneTable(mapFn PostgresMapFunc) PostgresSchema {
+	return PostgresSchema{
+		Tables: []PostgresTable{
+			{
+				Name: "events",
+				Columns: []PostgresColumn{
+					{Name: "event_id", Type: "TEXT", Nullable: false},
+					{Name: "emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
+					{Name: "payload_json", Type: "JSONB", Nullable: false},
+				},
+				PrimaryKey:  []string{"event_id"},
+				PruneColumn: "emitted_at",
+				Indexes: []PostgresIndex{
+					{Name: "idx_events_emitted_at", Columns: []string{"emitted_at"}},
+				},
+			},
+		},
+		MapEvent: mapFn,
+	}
+}
+
+func schemaTwoTables(mapFn PostgresMapFunc) PostgresSchema {
+	return PostgresSchema{
+		Tables: []PostgresTable{
+			{
+				Name: "events",
+				Columns: []PostgresColumn{
+					{Name: "event_id", Type: "TEXT", Nullable: false},
+					{Name: "emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
+				},
+				PrimaryKey:  []string{"event_id"},
+				PruneColumn: "emitted_at",
+			},
+			{
+				Name: "event_payloads",
+				Columns: []PostgresColumn{
+					{Name: "event_id", Type: "TEXT", Nullable: false},
+					{Name: "payload_json", Type: "JSONB", Nullable: false},
+					{Name: "emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
+				},
+				PrimaryKey:  []string{"event_id"},
+				PruneColumn: "emitted_at",
+			},
+		},
+		MapEvent: mapFn,
+	}
+}
+
+func mustCompileSchema(t *testing.T, s PostgresSchema) postgresSchemaCompiled {
+	t.Helper()
+	compiled, err := compilePostgresSchema(s)
+	if err != nil {
+		t.Fatalf("compile schema: %v", err)
+	}
+	return compiled
+}
+
+func TestRegisterPostgresSchema(t *testing.T) {
+	withPostgresTestState(t)
+
+	err := RegisterPostgresSchema("pg", schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+		return nil, nil
+	}))
+	if err != nil {
+		t.Fatalf("register schema: %v", err)
+	}
+
+	if _, ok := lookupPostgresSchema("pg"); !ok {
+		t.Fatalf("expected schema registration")
+	}
+
+	err = RegisterPostgresSchema("pg", schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+		return nil, nil
+	}))
+	if err == nil {
+		t.Fatalf("expected duplicate registration error")
+	}
+	if !strings.Contains(err.Error(), "already registered") {
+		t.Fatalf("unexpected duplicate error: %v", err)
+	}
+}
+
+func TestRegisterPostgresSchema_RejectsInvalidSchema(t *testing.T) {
+	withPostgresTestState(t)
+
+	err := RegisterPostgresSchema("pg", PostgresSchema{
+		Tables: []PostgresTable{
+			{
+				Name: "events",
+				Columns: []PostgresColumn{
+					{Name: "id", Type: "TEXT", Nullable: false},
+				},
+				PruneColumn: "missing_col",
+			},
+		},
+		MapEvent: func(_ context.Context, _ event.Event) ([]PostgresWrite, error) { return nil, nil },
+	})
+	if err == nil {
+		t.Fatalf("expected invalid schema error")
+	}
+	if !strings.Contains(err.Error(), "prune column") {
+		t.Fatalf("unexpected schema validation error: %v", err)
+	}
+
+	err = RegisterPostgresSchema("pg2", PostgresSchema{
+		Tables: []PostgresTable{
+			{
+				Name: "events",
+				Columns: []PostgresColumn{
+					{Name: "id", Type: "TEXT", Nullable: false},
+					{Name: "emitted_at", Type: "TIMESTAMPTZ", Nullable: false},
+				},
+				PruneColumn: "emitted_at",
+				Indexes: []PostgresIndex{
+					{Name: "idx_events_empty", Columns: nil},
+				},
+			},
+		},
+		MapEvent: func(_ context.Context, _ event.Event) ([]PostgresWrite, error) { return nil, nil },
+	})
+	if err == nil {
+		t.Fatalf("expected invalid index schema error")
+	}
+	if !strings.Contains(err.Error(), "at least one column") {
+		t.Fatalf("unexpected index validation error: %v", err)
+	}
+}
+
+func TestNewPostgresSinkFromConfig_MissingParams(t *testing.T) {
+	withPostgresTestState(t)
+
+	tests := []struct {
+		name   string
+		params map[string]any
+		want   string
+	}{
+		{name: "missing uri", params: map[string]any{"username": "u", "password": "p"}, want: "params.uri"},
+		{name: "missing username", params: map[string]any{"uri": "postgres://localhost/db", "password": "p"}, want: "params.username"},
+		{name: "missing password", params: map[string]any{"uri": "postgres://localhost/db", "username": "u"}, want: "params.password"},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := NewPostgresSinkFromConfig(config.SinkConfig{
+				Name:   "pg",
+				Driver: "postgres",
+				Params: tc.params,
+			})
+			if err == nil {
+				t.Fatalf("expected error")
+			}
+			if !strings.Contains(err.Error(), tc.want) {
+				t.Fatalf("expected %q in error, got: %v", tc.want, err)
+			}
+		})
+	}
+}
+
+func TestNewPostgresSinkFromConfig_MissingSchemaRegistration(t *testing.T) {
+	withPostgresTestState(t)
+
+	_, err := NewPostgresSinkFromConfig(config.SinkConfig{
+		Name:   "pg",
+		Driver: "postgres",
+		Params: map[string]any{
+			"uri":      "postgres://localhost/db",
+			"username": "user",
+			"password": "pass",
+		},
+	})
+	if err == nil {
+		t.Fatalf("expected error")
+	}
+	if !strings.Contains(err.Error(), "no schema registered") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestNewPostgresSinkFromConfig_EagerInit(t *testing.T) {
+	withPostgresTestState(t)
+
+	err := RegisterPostgresSchema("pg", schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+		return nil, nil
+	}))
+	if err != nil {
+		t.Fatalf("register schema: %v", err)
+	}
+
+	db := &fakeDB{}
+	var gotDSN string
+	openPostgresDB = func(dsn string) (postgresDB, error) {
+		gotDSN = dsn
+		return db, nil
+	}
+
+	s, err := NewPostgresSinkFromConfig(config.SinkConfig{
+		Name:   "pg",
+		Driver: "postgres",
+		Params: map[string]any{
+			"uri":      "postgres://db.example.local:5432/feedkit?sslmode=disable",
+			"username": "app_user",
+			"password": "app_pass",
+		},
+	})
+	if err != nil {
+		t.Fatalf("new postgres sink: %v", err)
+	}
+	if s == nil {
+		t.Fatalf("expected sink")
+	}
+
+	if db.pingCalls != 1 {
+		t.Fatalf("expected one ping, got %d", db.pingCalls)
+	}
+	if len(db.execCalls) != 2 {
+		t.Fatalf("expected 2 init exec calls (table + index), got %d", len(db.execCalls))
+	}
+	if !strings.Contains(db.execCalls[0].query, `CREATE TABLE IF NOT EXISTS "events"`) {
+		t.Fatalf("unexpected create table query: %s", db.execCalls[0].query)
+	}
+	if !strings.Contains(db.execCalls[1].query, `CREATE INDEX IF NOT EXISTS "idx_events_emitted_at"`) {
+		t.Fatalf("unexpected create index query: %s", db.execCalls[1].query)
+	}
+
+	u, err := url.Parse(gotDSN)
+	if err != nil {
+		t.Fatalf("parse dsn: %v", err)
+	}
+	if u.User == nil || u.User.Username() != "app_user" {
+		t.Fatalf("dsn missing username: %q", gotDSN)
+	}
+	pass, ok := u.User.Password()
+	if !ok || pass != "app_pass" {
+		t.Fatalf("dsn missing password: %q", gotDSN)
+	}
+}
+
+func TestNewPostgresSinkFromConfig_InitFailureClosesDB(t *testing.T) {
+	withPostgresTestState(t)
+
+	err := RegisterPostgresSchema("pg", schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+		return nil, nil
+	}))
+	if err != nil {
+		t.Fatalf("register schema: %v", err)
+	}
+
+	db := &fakeDB{execErrOnCall: 1, execErr: errors.New("ddl failed")}
+	openPostgresDB = func(_ string) (postgresDB, error) {
+		return db, nil
+	}
+
+	_, err = NewPostgresSinkFromConfig(config.SinkConfig{
+		Name:   "pg",
+		Driver: "postgres",
+		Params: map[string]any{
+			"uri":      "postgres://localhost/db",
+			"username": "user",
+			"password": "pass",
+		},
+	})
+	if err == nil {
+		t.Fatalf("expected init error")
+	}
+	if db.closeCalls != 1 {
+		t.Fatalf("expected db close on init failure")
+	}
+}
+
+func TestNewPostgresSinkFromConfig_PruneParamAccepted(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want time.Duration
+	}{
+		{name: "go duration", in: "72h", want: 72 * time.Hour},
+		{name: "days suffix", in: "3d", want: 72 * time.Hour},
+		{name: "weeks suffix", in: "2w", want: 14 * 24 * time.Hour},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			withPostgresTestState(t)
+
+			err := RegisterPostgresSchema("pg", schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+				return nil, nil
+			}))
+			if err != nil {
+				t.Fatalf("register schema: %v", err)
+			}
+
+			openPostgresDB = func(_ string) (postgresDB, error) {
+				return &fakeDB{}, nil
+			}
+
+			s, err := NewPostgresSinkFromConfig(config.SinkConfig{
+				Name:   "pg",
+				Driver: "postgres",
+				Params: map[string]any{
+					"uri":      "postgres://localhost/db",
+					"username": "user",
+					"password": "pass",
+					"prune":    tc.in,
+				},
+			})
+			if err != nil {
+				t.Fatalf("new postgres sink: %v", err)
+			}
+
+			pg, ok := s.(*PostgresSink)
+			if !ok {
+				t.Fatalf("expected *PostgresSink, got %T", s)
+			}
+			if pg.pruneWindow != tc.want {
+				t.Fatalf("prune window = %s, want %s", pg.pruneWindow, tc.want)
+			}
+		})
+	}
+}
+
+func TestNewPostgresSinkFromConfig_PruneParamRejected(t *testing.T) {
+	withPostgresTestState(t)
+
+	tests := []struct {
+		name string
+		in   any
+	}{
+		{name: "empty", in: ""},
+		{name: "zero", in: "0"},
+		{name: "negative", in: "-1h"},
+		{name: "malformed", in: "abc"},
+		{name: "fractional day", in: "1.5d"},
+		{name: "wrong type", in: 5},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := NewPostgresSinkFromConfig(config.SinkConfig{
+				Name:   "pg",
+				Driver: "postgres",
+				Params: map[string]any{
+					"uri":      "postgres://localhost/db",
+					"username": "user",
+					"password": "pass",
+					"prune":    tc.in,
+				},
+			})
+			if err == nil {
+				t.Fatalf("expected error")
+			}
+			if !strings.Contains(err.Error(), "params.prune") {
+				t.Fatalf("expected params.prune error, got %v", err)
+			}
+		})
+	}
+}
+
+func TestPostgresSinkConsume_InvalidEvent(t *testing.T) {
+	db := &fakeDB{}
+	called := 0
+	sink := &PostgresSink{
+		name: "pg",
+		db:   db,
+		schema: mustCompileSchema(t, schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+			called++
+			return nil, nil
+		})),
+	}
+
+	err := sink.Consume(context.Background(), event.Event{})
+	if err == nil {
+		t.Fatalf("expected invalid event error")
+	}
+	if !strings.Contains(err.Error(), "invalid event") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if called != 0 {
+		t.Fatalf("expected mapper not called for invalid events")
+	}
+}
+
+func TestPostgresSinkConsume_UnmappedEventIsNoOp(t *testing.T) {
+	db := &fakeDB{}
+	sink := &PostgresSink{
+		name: "pg",
+		db:   db,
+		schema: mustCompileSchema(t, schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+			return nil, nil
+		})),
+	}
+
+	if err := sink.Consume(context.Background(), validTestEvent()); err != nil {
+		t.Fatalf("consume: %v", err)
+	}
+	if db.beginCalls != 0 {
+		t.Fatalf("expected no transaction for unmapped events")
+	}
+}
+
+func TestPostgresSinkConsume_OneEventWritesMultipleTablesAtomically(t *testing.T) {
+	tx := &fakeTx{}
+	db := &fakeDB{tx: tx}
+	sink := &PostgresSink{
+		name: "pg",
+		db:   db,
+		schema: mustCompileSchema(t, schemaTwoTables(func(_ context.Context, e event.Event) ([]PostgresWrite, error) {
+			return []PostgresWrite{
+				{Table: "events", Values: map[string]any{"event_id": e.ID, "emitted_at": e.EmittedAt}},
+				{Table: "event_payloads", Values: map[string]any{"event_id": e.ID, "payload_json": `{}`, "emitted_at": e.EmittedAt}},
+			}, nil
+		})),
+	}
+
+	if err := sink.Consume(context.Background(), validTestEvent()); err != nil {
+		t.Fatalf("consume: %v", err)
+	}
+	if db.beginCalls != 1 {
+		t.Fatalf("expected one transaction begin, got %d", db.beginCalls)
+	}
+	if len(tx.execCalls) != 2 {
+		t.Fatalf("expected 2 insert statements, got %d", len(tx.execCalls))
+	}
+	if tx.commitCalls != 1 {
+		t.Fatalf("expected one commit, got %d", tx.commitCalls)
+	}
+	if tx.rollbackCalls != 0 {
+		t.Fatalf("expected zero rollbacks, got %d", tx.rollbackCalls)
+	}
+}
+
+func TestPostgresSinkConsume_InsertFailureRollsBack(t *testing.T) {
+	tx := &fakeTx{execErrOnCall: 2, execErr: errors.New("duplicate key")}
+	db := &fakeDB{tx: tx}
+	sink := &PostgresSink{
+		name: "pg",
+		db:   db,
+		schema: mustCompileSchema(t, schemaTwoTables(func(_ context.Context, e event.Event) ([]PostgresWrite, error) {
+			return []PostgresWrite{
+				{Table: "events", Values: map[string]any{"event_id": e.ID, "emitted_at": e.EmittedAt}},
+				{Table: "event_payloads", Values: map[string]any{"event_id": e.ID, "payload_json": `{}`, "emitted_at": e.EmittedAt}},
+			}, nil
+		})),
+	}
+
+	err := sink.Consume(context.Background(), validTestEvent())
+	if err == nil {
+		t.Fatalf("expected insert error")
+	}
+	if !strings.Contains(err.Error(), "insert into") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if tx.commitCalls != 0 {
+		t.Fatalf("expected no commit")
+	}
+	if tx.rollbackCalls != 1 {
+		t.Fatalf("expected rollback, got %d", tx.rollbackCalls)
+	}
+}
+
+func TestPostgresSinkConsume_AutoPruneRunsInSameTransaction(t *testing.T) {
+	tx := &fakeTx{}
+	db := &fakeDB{tx: tx}
+	sink := &PostgresSink{
+		name:        "pg",
+		db:          db,
+		schema:      mustCompileSchema(t, schemaTwoTables(func(_ context.Context, e event.Event) ([]PostgresWrite, error) {
+			return []PostgresWrite{
+				{Table: "events", Values: map[string]any{"event_id": e.ID, "emitted_at": e.EmittedAt}},
+				{Table: "event_payloads", Values: map[string]any{"event_id": e.ID, "payload_json": `{}`, "emitted_at": e.EmittedAt}},
+			}, nil
+		})),
+		pruneWindow: 24 * time.Hour,
+	}
+
+	if err := sink.Consume(context.Background(), validTestEvent()); err != nil {
+		t.Fatalf("consume: %v", err)
+	}
+	if len(tx.execCalls) != 4 {
+		t.Fatalf("expected 4 tx statements (2 inserts + 2 prunes), got %d", len(tx.execCalls))
+	}
+	if !strings.Contains(tx.execCalls[2].query, `DELETE FROM "events"`) {
+		t.Fatalf("expected prune delete for events, got %s", tx.execCalls[2].query)
+	}
+	if !strings.Contains(tx.execCalls[3].query, `DELETE FROM "event_payloads"`) {
+		t.Fatalf("expected prune delete for event_payloads, got %s", tx.execCalls[3].query)
+	}
+	if tx.commitCalls != 1 {
+		t.Fatalf("expected one commit, got %d", tx.commitCalls)
+	}
+	if tx.rollbackCalls != 0 {
+		t.Fatalf("expected zero rollbacks, got %d", tx.rollbackCalls)
+	}
+}
+
+func TestPostgresSinkConsume_AutoPruneFailureRollsBack(t *testing.T) {
+	tx := &fakeTx{execErrOnCall: 3, execErr: errors.New("prune failed")}
+	db := &fakeDB{tx: tx}
+	sink := &PostgresSink{
+		name:        "pg",
+		db:          db,
+		schema:      mustCompileSchema(t, schemaTwoTables(func(_ context.Context, e event.Event) ([]PostgresWrite, error) {
+			return []PostgresWrite{
+				{Table: "events", Values: map[string]any{"event_id": e.ID, "emitted_at": e.EmittedAt}},
+				{Table: "event_payloads", Values: map[string]any{"event_id": e.ID, "payload_json": `{}`, "emitted_at": e.EmittedAt}},
+			}, nil
+		})),
+		pruneWindow: 24 * time.Hour,
+	}
+
+	err := sink.Consume(context.Background(), validTestEvent())
+	if err == nil {
+		t.Fatalf("expected prune error")
+	}
+	if !strings.Contains(err.Error(), "prune older than") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if tx.commitCalls != 0 {
+		t.Fatalf("expected no commit")
+	}
+	if tx.rollbackCalls != 1 {
+		t.Fatalf("expected rollback, got %d", tx.rollbackCalls)
+	}
+}
+
+func TestPostgresSinkPrune_PerTable(t *testing.T) {
+	db := &fakeDB{execRows: 7}
+	sink := &PostgresSink{
+		name: "pg",
+		db:   db,
+		schema: mustCompileSchema(t, schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+			return nil, nil
+		})),
+	}
+
+	rows, err := sink.PruneKeepLatest(context.Background(), "events", 10)
+	if err != nil {
+		t.Fatalf("prune keep latest: %v", err)
+	}
+	if rows != 7 {
+		t.Fatalf("unexpected rows affected: %d", rows)
+	}
+	if len(db.execCalls) != 1 {
+		t.Fatalf("expected one prune query")
+	}
+	if !strings.Contains(db.execCalls[0].query, `ORDER BY "emitted_at" DESC`) {
+		t.Fatalf("unexpected keep-latest query: %s", db.execCalls[0].query)
+	}
+	if len(db.execCalls[0].args) != 1 || db.execCalls[0].args[0] != 10 {
+		t.Fatalf("unexpected keep-latest args: %#v", db.execCalls[0].args)
+	}
+
+	cutoff := time.Now().UTC().Add(-24 * time.Hour)
+	rows, err = sink.PruneOlderThan(context.Background(), "events", cutoff)
+	if err != nil {
+		t.Fatalf("prune older than: %v", err)
+	}
+	if rows != 7 {
+		t.Fatalf("unexpected rows affected: %d", rows)
+	}
+	if len(db.execCalls) != 2 {
+		t.Fatalf("expected two prune queries")
+	}
+	if !strings.Contains(db.execCalls[1].query, `WHERE "emitted_at" < $1`) {
+		t.Fatalf("unexpected older-than query: %s", db.execCalls[1].query)
+	}
+}
+
+func TestPostgresSinkPrune_AllTables(t *testing.T) {
+	db := &fakeDB{execRows: 3}
+	sink := &PostgresSink{
+		name: "pg",
+		db:   db,
+		schema: mustCompileSchema(t, schemaTwoTables(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+			return nil, nil
+		})),
+	}
+
+	keepCounts, err := sink.PruneAllKeepLatest(context.Background(), 5)
+	if err != nil {
+		t.Fatalf("prune all keep latest: %v", err)
+	}
+	if len(keepCounts) != 2 || keepCounts["events"] != 3 || keepCounts["event_payloads"] != 3 {
+		t.Fatalf("unexpected keep counts: %#v", keepCounts)
+	}
+
+	db.execCalls = nil
+	olderCounts, err := sink.PruneAllOlderThan(context.Background(), time.Now().UTC())
+	if err != nil {
+		t.Fatalf("prune all older than: %v", err)
+	}
+	if len(olderCounts) != 2 || olderCounts["events"] != 3 || olderCounts["event_payloads"] != 3 {
+		t.Fatalf("unexpected older-than counts: %#v", olderCounts)
+	}
+	if len(db.execCalls) != 2 {
+		t.Fatalf("expected one prune call per table")
+	}
+}
+
+func TestPostgresSinkPrune_Errors(t *testing.T) {
+	db := &fakeDB{}
+	sink := &PostgresSink{
+		name: "pg",
+		db:   db,
+		schema: mustCompileSchema(t, schemaOneTable(func(_ context.Context, _ event.Event) ([]PostgresWrite, error) {
+			return nil, nil
+		})),
+	}
+
+	if _, err := sink.PruneKeepLatest(context.Background(), "events", -1); err == nil {
+		t.Fatalf("expected negative keep error")
+	}
+	if _, err := sink.PruneOlderThan(context.Background(), "missing", time.Now().UTC()); err == nil {
+		t.Fatalf("expected unknown table error")
+	}
+}

sources/doc.go

@@ -11,4 +11,14 @@
 //
 // A single source may emit 0..N events per poll or stream iteration, and those
 // events may span multiple event kinds.
+//
+// HTTP-backed polling sources can share NewHTTPSource for generic HTTP config
+// parsing and conditional GET behavior. The helper understands:
+//   - params.url
+//   - params.user_agent
+//   - params.conditional (optional, default true)
+//
+// When validators are available, NewHTTPSource prefers ETag/If-None-Match and
+// falls back to Last-Modified/If-Modified-Since. A 304 Not Modified response is
+// treated as a successful unchanged poll.
 package sources

sources/helpers.go

@@ -0,0 +1,145 @@
+package sources
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+// DefaultEventID applies feedkit's default Event.ID policy:
+//
+//   - If upstream provides an ID, use it (trimmed).
+//   - Otherwise, ID is "<Source>:<EffectiveAt>" when available.
+//   - If EffectiveAt is unavailable, fall back to "<Source>:<EmittedAt>".
+//
+// Timestamps are encoded as RFC3339Nano in UTC.
+func DefaultEventID(upstreamID, sourceName string, effectiveAt *time.Time, emittedAt time.Time) string {
+	if id := strings.TrimSpace(upstreamID); id != "" {
+		return id
+	}
+
+	src := strings.TrimSpace(sourceName)
+	if src == "" {
+		src = "UNKNOWN_SOURCE"
+	}
+
+	if effectiveAt != nil && !effectiveAt.IsZero() {
+		return fmt.Sprintf("%s:%s", src, effectiveAt.UTC().Format(time.RFC3339Nano))
+	}
+
+	t := emittedAt.UTC()
+	if t.IsZero() {
+		t = time.Now().UTC()
+	}
+
+	return fmt.Sprintf("%s:%s", src, t.Format(time.RFC3339Nano))
+}
+
+// SingleEvent constructs, validates, and returns a slice containing exactly one event.
+func SingleEvent(
+	kind event.Kind,
+	sourceName string,
+	schema string,
+	id string,
+	emittedAt time.Time,
+	effectiveAt *time.Time,
+	payload any,
+) ([]event.Event, error) {
+	if emittedAt.IsZero() {
+		emittedAt = time.Now().UTC()
+	} else {
+		emittedAt = emittedAt.UTC()
+	}
+
+	e := event.Event{
+		ID:          id,
+		Kind:        kind,
+		Source:      sourceName,
+		EmittedAt:   emittedAt,
+		EffectiveAt: effectiveAt,
+		Schema:      schema,
+		Payload:     payload,
+	}
+
+	if err := e.Validate(); err != nil {
+		return nil, err
+	}
+
+	return []event.Event{e}, nil
+}
+
+// ValidateExpectedKinds checks that configured source expected kinds are a subset
+// of the kinds advertised by the built source, when the source exposes kind
+// metadata. If the source does not advertise kinds, the check is skipped.
+func ValidateExpectedKinds(cfg config.SourceConfig, in Input) error {
+	expectedKinds, err := parseExpectedKinds(cfg.ExpectedKinds())
+	if err != nil {
+		return err
+	}
+	if len(expectedKinds) == 0 {
+		return nil
+	}
+
+	advertisedKinds := advertisedSourceKinds(in)
+	if len(advertisedKinds) == 0 {
+		return nil
+	}
+
+	for kind := range expectedKinds {
+		if !advertisedKinds[kind] {
+			return fmt.Errorf(
+				"configured expected kind %q not advertised by source (configured=%v advertised=%v)",
+				kind,
+				sortedKinds(expectedKinds),
+				sortedKinds(advertisedKinds),
+			)
+		}
+	}
+	return nil
+}
+
+func parseExpectedKinds(raw []string) (map[event.Kind]bool, error) {
+	kinds := map[event.Kind]bool{}
+	for i, k := range raw {
+		kind, err := event.ParseKind(k)
+		if err != nil {
+			return nil, fmt.Errorf("invalid expected kind at index %d (%q): %w", i, k, err)
+		}
+		kinds[kind] = true
+	}
+	return kinds, nil
+}
+
+func advertisedSourceKinds(in Input) map[event.Kind]bool {
+	if in == nil {
+		return nil
+	}
+
+	kinds := map[event.Kind]bool{}
+	if ks, ok := in.(KindsSource); ok {
+		for _, kind := range ks.Kinds() {
+			kinds[kind] = true
+		}
+		return kinds
+	}
+
+	if ks, ok := in.(KindSource); ok {
+		kinds[ks.Kind()] = true
+		return kinds
+	}
+
+	return nil
+}
+
+func sortedKinds(kindSet map[event.Kind]bool) []string {
+	out := make([]string, 0, len(kindSet))
+	for kind := range kindSet {
+		out = append(out, string(kind))
+	}
+	sort.Strings(out)
+	return out
+}

sources/helpers_test.go

@@ -0,0 +1,131 @@
+package sources
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+type testInput struct {
+	name string
+}
+
+func (s testInput) Name() string { return s.name }
+
+type testKindSource struct {
+	testInput
+	kind event.Kind
+}
+
+func (s testKindSource) Kind() event.Kind { return s.kind }
+
+type testKindsSource struct {
+	testInput
+	kinds []event.Kind
+}
+
+func (s testKindsSource) Kinds() []event.Kind { return s.kinds }
+
+func TestValidateExpectedKindsLegacyKindFallback(t *testing.T) {
+	cfg := config.SourceConfig{Kind: "observation"}
+	in := testKindSource{
+		testInput: testInput{name: "test"},
+		kind:      event.Kind("observation"),
+	}
+
+	if err := ValidateExpectedKinds(cfg, in); err != nil {
+		t.Fatalf("ValidateExpectedKinds() unexpected error: %v", err)
+	}
+}
+
+func TestValidateExpectedKindsSubsetAllowed(t *testing.T) {
+	cfg := config.SourceConfig{Kinds: []string{"observation"}}
+	in := testKindsSource{
+		testInput: testInput{name: "test"},
+		kinds:     []event.Kind{"observation", "forecast"},
+	}
+
+	if err := ValidateExpectedKinds(cfg, in); err != nil {
+		t.Fatalf("ValidateExpectedKinds() unexpected error: %v", err)
+	}
+}
+
+func TestValidateExpectedKindsMismatchFails(t *testing.T) {
+	cfg := config.SourceConfig{Kinds: []string{"alert"}}
+	in := testKindsSource{
+		testInput: testInput{name: "test"},
+		kinds:     []event.Kind{"observation", "forecast"},
+	}
+
+	err := ValidateExpectedKinds(cfg, in)
+	if err == nil {
+		t.Fatalf("ValidateExpectedKinds() expected mismatch error, got nil")
+	}
+	if !strings.Contains(err.Error(), "configured expected kind") {
+		t.Fatalf("ValidateExpectedKinds() error %q does not include expected message", err)
+	}
+}
+
+func TestValidateExpectedKindsNoMetadataSkipsCheck(t *testing.T) {
+	cfg := config.SourceConfig{Kinds: []string{"alert"}}
+	in := testInput{name: "test"}
+
+	if err := ValidateExpectedKinds(cfg, in); err != nil {
+		t.Fatalf("ValidateExpectedKinds() unexpected error: %v", err)
+	}
+}
+
+func TestDefaultEventIDUsesUpstreamID(t *testing.T) {
+	emittedAt := time.Date(2026, 3, 28, 15, 4, 5, 123, time.UTC)
+	got := DefaultEventID(" upstream-id ", "source", nil, emittedAt)
+	if got != "upstream-id" {
+		t.Fatalf("DefaultEventID() = %q, want upstream-id", got)
+	}
+}
+
+func TestDefaultEventIDPrefersEffectiveAt(t *testing.T) {
+	effectiveAt := time.Date(2026, 3, 28, 16, 4, 5, 987654321, time.FixedZone("x", -6*3600))
+	emittedAt := time.Date(2026, 3, 28, 15, 4, 5, 123, time.UTC)
+
+	got := DefaultEventID("", "source", &effectiveAt, emittedAt)
+	want := "source:" + effectiveAt.UTC().Format(time.RFC3339Nano)
+	if got != want {
+		t.Fatalf("DefaultEventID() = %q, want %q", got, want)
+	}
+}
+
+func TestDefaultEventIDFallsBackToEmittedAt(t *testing.T) {
+	emittedAt := time.Date(2026, 3, 28, 15, 4, 5, 123456789, time.FixedZone("y", 3*3600))
+	got := DefaultEventID("", "source", nil, emittedAt)
+	want := "source:" + emittedAt.UTC().Format(time.RFC3339Nano)
+	if got != want {
+		t.Fatalf("DefaultEventID() = %q, want %q", got, want)
+	}
+}
+
+func TestSingleEventBuildsValidatedSlice(t *testing.T) {
+	effectiveAt := time.Date(2026, 3, 28, 16, 0, 0, 0, time.UTC)
+	emittedAt := time.Date(2026, 3, 28, 15, 0, 0, 0, time.FixedZone("z", -5*3600))
+
+	got, err := SingleEvent(
+		event.Kind("observation"),
+		"source-a",
+		"raw.example.v1",
+		"evt-1",
+		emittedAt,
+		&effectiveAt,
+		map[string]any{"ok": true},
+	)
+	if err != nil {
+		t.Fatalf("SingleEvent() unexpected error: %v", err)
+	}
+	if len(got) != 1 {
+		t.Fatalf("SingleEvent() len = %d, want 1", len(got))
+	}
+	if got[0].EmittedAt != emittedAt.UTC() {
+		t.Fatalf("SingleEvent() emittedAt = %s, want %s", got[0].EmittedAt, emittedAt.UTC())
+	}
+}

sources/http.go

@@ -0,0 +1,147 @@
+package sources
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+	"gitea.maximumdirect.net/ejr/feedkit/transport"
+)
+
+// HTTPSource is a reusable helper for polling HTTP-backed sources.
+//
+// It centralizes generic source config parsing (`params.url`,
+// `params.user_agent`, and optional `params.conditional`), default HTTP client
+// setup, and conditional GET validator handling. Concrete daemon sources remain
+// responsible for decoding the response body and constructing events.
+type HTTPSource struct {
+	Driver      string
+	Name        string
+	URL         string
+	UserAgent   string
+	Accept      string
+	Conditional bool
+	Client      *http.Client
+
+	mu         sync.Mutex
+	validators transport.HTTPValidators
+}
+
+// NewHTTPSource builds a generic HTTP polling helper from SourceConfig.
+//
+// Required params:
+//   - params.url
+//   - params.user_agent
+//
+// Optional params:
+//   - params.conditional (default true): enable conditional GET using cached
+//     ETag / Last-Modified validators
+func NewHTTPSource(driver string, cfg config.SourceConfig, accept string) (*HTTPSource, error) {
+	name := strings.TrimSpace(cfg.Name)
+	if name == "" {
+		return nil, fmt.Errorf("%s: name is required", driver)
+	}
+	if cfg.Params == nil {
+		return nil, fmt.Errorf("%s %q: params are required (need params.url and params.user_agent)", driver, cfg.Name)
+	}
+
+	url, ok := cfg.ParamString("url", "URL")
+	if !ok {
+		return nil, fmt.Errorf("%s %q: params.url is required", driver, cfg.Name)
+	}
+
+	userAgent, ok := cfg.ParamString("user_agent", "userAgent")
+	if !ok {
+		return nil, fmt.Errorf("%s %q: params.user_agent is required", driver, cfg.Name)
+	}
+
+	conditional, err := parseConditionalParam(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return &HTTPSource{
+		Driver:      driver,
+		Name:        name,
+		URL:         url,
+		UserAgent:   userAgent,
+		Accept:      accept,
+		Conditional: conditional,
+		Client:      transport.NewHTTPClient(transport.DefaultHTTPTimeout),
+	}, nil
+}
+
+// FetchBytesIfChanged fetches the configured URL and reports whether the
+// upstream content changed. An unchanged 304 response returns changed=false
+// with no body and no error.
+func (s *HTTPSource) FetchBytesIfChanged(ctx context.Context) ([]byte, bool, error) {
+	client := s.Client
+	if client == nil {
+		client = transport.NewHTTPClient(transport.DefaultHTTPTimeout)
+	}
+
+	s.mu.Lock()
+	validators := s.validators
+	s.mu.Unlock()
+
+	body, changed, next, err := transport.FetchBodyIfChanged(
+		ctx,
+		client,
+		s.URL,
+		s.UserAgent,
+		s.Accept,
+		s.Conditional,
+		validators,
+	)
+	if err != nil {
+		return nil, false, fmt.Errorf("%s %q: %w", s.Driver, s.Name, err)
+	}
+
+	if s.Conditional {
+		s.mu.Lock()
+		s.validators = next
+		s.mu.Unlock()
+	}
+
+	return body, changed, nil
+}
+
+// FetchJSONIfChanged fetches the configured URL and returns the raw response
+// body as json.RawMessage when content changed. An unchanged 304 response
+// returns changed=false with a nil body and no error.
+func (s *HTTPSource) FetchJSONIfChanged(ctx context.Context) (json.RawMessage, bool, error) {
+	body, changed, err := s.FetchBytesIfChanged(ctx)
+	if err != nil || !changed {
+		return nil, changed, err
+	}
+	return json.RawMessage(body), true, nil
+}
+
+func parseConditionalParam(cfg config.SourceConfig) (bool, error) {
+	raw, ok := cfg.Params["conditional"]
+	if !ok || raw == nil {
+		return true, nil
+	}
+
+	switch v := raw.(type) {
+	case bool:
+		return v, nil
+	case string:
+		s := strings.TrimSpace(v)
+		if s == "" {
+			return false, fmt.Errorf("source %q: params.conditional must be a boolean", cfg.Name)
+		}
+		parsed, err := strconv.ParseBool(s)
+		if err != nil {
+			return false, fmt.Errorf("source %q: params.conditional must be a boolean", cfg.Name)
+		}
+		return parsed, nil
+	default:
+		return false, fmt.Errorf("source %q: params.conditional must be a boolean", cfg.Name)
+	}
+}

sources/http_test.go

@@ -0,0 +1,96 @@
+package sources
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"gitea.maximumdirect.net/ejr/feedkit/config"
+)
+
+func TestNewHTTPSourceConditionalDefaultsTrue(t *testing.T) {
+	src, err := NewHTTPSource("test_driver", config.SourceConfig{
+		Name:   "test-source",
+		Driver: "test_driver",
+		Params: map[string]any{
+			"url":        "https://example.invalid",
+			"user_agent": "test-agent",
+		},
+	}, "application/json")
+	if err != nil {
+		t.Fatalf("NewHTTPSource() error = %v", err)
+	}
+	if !src.Conditional {
+		t.Fatalf("Conditional = false, want true")
+	}
+}
+
+func TestNewHTTPSourceRejectsInvalidConditional(t *testing.T) {
+	_, err := NewHTTPSource("test_driver", config.SourceConfig{
+		Name:   "test-source",
+		Driver: "test_driver",
+		Params: map[string]any{
+			"url":         "https://example.invalid",
+			"user_agent":  "test-agent",
+			"conditional": "sometimes",
+		},
+	}, "application/json")
+	if err == nil {
+		t.Fatalf("NewHTTPSource() error = nil, want error")
+	}
+}
+
+func TestHTTPSourceFetchJSONIfChanged(t *testing.T) {
+	var call int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		call++
+		switch call {
+		case 1:
+			w.Header().Set("ETag", `"v1"`)
+			_, _ = w.Write([]byte(`{"ok":true}`))
+		case 2:
+			if got := r.Header.Get("If-None-Match"); got != `"v1"` {
+				t.Fatalf("second request If-None-Match = %q", got)
+			}
+			w.WriteHeader(http.StatusNotModified)
+		default:
+			t.Fatalf("unexpected call count %d", call)
+		}
+	}))
+	defer srv.Close()
+
+	src, err := NewHTTPSource("test_driver", config.SourceConfig{
+		Name:   "test-source",
+		Driver: "test_driver",
+		Params: map[string]any{
+			"url":        srv.URL,
+			"user_agent": "test-agent",
+		},
+	}, "application/json")
+	if err != nil {
+		t.Fatalf("NewHTTPSource() error = %v", err)
+	}
+
+	raw, changed, err := src.FetchJSONIfChanged(context.Background())
+	if err != nil {
+		t.Fatalf("first FetchJSONIfChanged() error = %v", err)
+	}
+	if !changed {
+		t.Fatalf("first FetchJSONIfChanged() changed = false, want true")
+	}
+	if got := string(raw); got != `{"ok":true}` {
+		t.Fatalf("first FetchJSONIfChanged() body = %q", got)
+	}
+
+	raw, changed, err = src.FetchJSONIfChanged(context.Background())
+	if err != nil {
+		t.Fatalf("second FetchJSONIfChanged() error = %v", err)
+	}
+	if changed {
+		t.Fatalf("second FetchJSONIfChanged() changed = true, want false")
+	}
+	if raw != nil {
+		t.Fatalf("second FetchJSONIfChanged() body = %q, want nil", string(raw))
+	}
+}

transport/http.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"strings"
 	"time"
 )
 
@@ -28,7 +29,80 @@ func NewHTTPClient(timeout time.Duration) *http.Client {
 }
 
 func FetchBody(ctx context.Context, client *http.Client, url, userAgent, accept string) ([]byte, error) {
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	res, err := doRequest(ctx, client, http.MethodGet, url, userAgent, accept, "", "")
+	if err != nil {
+		return nil, err
+	}
+	defer res.Body.Close()
+
+	if res.StatusCode < 200 || res.StatusCode >= 300 {
+		return nil, fmt.Errorf("HTTP %s", res.Status)
+	}
+
+	return readValidatedBody(res.Body)
+}
+
+// HTTPValidators are cache validators learned from prior successful GET responses.
+//
+// ETag is preferred when present. LastModified is used as a fallback validator
+// when ETag is unavailable.
+type HTTPValidators struct {
+	ETag         string
+	LastModified string
+}
+
+// FetchBodyIfChanged performs an HTTP GET and opportunistically uses conditional
+// request headers based on the provided validators.
+//
+// Behavior:
+//   - if conditional is false, this behaves like a normal GET and leaves validators unchanged
+//   - if validators.ETag is set, sends If-None-Match
+//   - else if validators.LastModified is set, sends If-Modified-Since
+//   - 304 Not Modified is treated as success with changed=false and no body
+//   - 200 responses are treated as changed=true and still enforce the normal body checks
+//
+// Returned validators reflect any updates learned from the response headers.
+func FetchBodyIfChanged(
+	ctx context.Context,
+	client *http.Client,
+	url, userAgent, accept string,
+	conditional bool,
+	validators HTTPValidators,
+) ([]byte, bool, HTTPValidators, error) {
+	headerName, headerValue := conditionalHeader(conditional, validators)
+
+	res, err := doRequest(ctx, client, http.MethodGet, url, userAgent, accept, headerName, headerValue)
+	if err != nil {
+		return nil, false, validators, err
+	}
+	defer res.Body.Close()
+
+	switch res.StatusCode {
+	case http.StatusNotModified:
+		if conditional {
+			validators = refreshValidators(validators, res.Header)
+		}
+		return nil, false, validators, nil
+	default:
+		if res.StatusCode < 200 || res.StatusCode >= 300 {
+			return nil, false, validators, fmt.Errorf("HTTP %s", res.Status)
+		}
+	}
+
+	b, err := readValidatedBody(res.Body)
+	if err != nil {
+		return nil, false, validators, err
+	}
+
+	if conditional {
+		validators = replaceValidators(res.Header)
+	}
+
+	return b, true, validators, nil
+}
+
+func doRequest(ctx context.Context, client *http.Client, method, url, userAgent, accept, headerName, headerValue string) (*http.Response, error) {
+	req, err := http.NewRequestWithContext(ctx, method, url, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -39,19 +113,46 @@ func FetchBody(ctx context.Context, client *http.Client, url, userAgent, accept
 	if accept != "" {
 		req.Header.Set("Accept", accept)
 	}
-
+	if headerName != "" && headerValue != "" {
-	res, err := client.Do(req)
+		req.Header.Set(headerName, headerValue)
-	if err != nil {
-		return nil, err
-	}
-	defer res.Body.Close()
-
-	if res.StatusCode < 200 || res.StatusCode >= 300 {
-		return nil, fmt.Errorf("HTTP %s", res.Status)
 	}
 
+	return client.Do(req)
+}
+
+func conditionalHeader(enabled bool, validators HTTPValidators) (string, string) {
+	if !enabled {
+		return "", ""
+	}
+	if etag := strings.TrimSpace(validators.ETag); etag != "" {
+		return "If-None-Match", etag
+	}
+	if lastModified := strings.TrimSpace(validators.LastModified); lastModified != "" {
+		return "If-Modified-Since", lastModified
+	}
+	return "", ""
+}
+
+func replaceValidators(header http.Header) HTTPValidators {
+	return HTTPValidators{
+		ETag:         strings.TrimSpace(header.Get("ETag")),
+		LastModified: strings.TrimSpace(header.Get("Last-Modified")),
+	}
+}
+
+func refreshValidators(current HTTPValidators, header http.Header) HTTPValidators {
+	if etag := strings.TrimSpace(header.Get("ETag")); etag != "" {
+		current.ETag = etag
+	}
+	if lastModified := strings.TrimSpace(header.Get("Last-Modified")); lastModified != "" {
+		current.LastModified = lastModified
+	}
+	return current
+}
+
+func readValidatedBody(r io.Reader) ([]byte, error) {
 	// Read at most maxResponseBodyBytes + 1 so we can detect overflow.
-	limited := io.LimitReader(res.Body, maxResponseBodyBytes+1)
+	limited := io.LimitReader(r, maxResponseBodyBytes+1)
 
 	b, err := io.ReadAll(limited)
 	if err != nil {

transport/http_test.go

@@ -0,0 +1,232 @@
+package transport
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+)
+
+func TestFetchBodyIfChangedPrefersETagAndTreats304AsUnchanged(t *testing.T) {
+	t.Helper()
+
+	var call int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		call++
+		switch call {
+		case 1:
+			if got := r.Header.Get("If-None-Match"); got != "" {
+				t.Fatalf("first request If-None-Match = %q, want empty", got)
+			}
+			if got := r.Header.Get("If-Modified-Since"); got != "" {
+				t.Fatalf("first request If-Modified-Since = %q, want empty", got)
+			}
+			w.Header().Set("ETag", `"v1"`)
+			w.Header().Set("Last-Modified", "Mon, 02 Jan 2006 15:04:05 GMT")
+			_, _ = w.Write([]byte(`{"ok":true}`))
+		case 2:
+			if got := r.Header.Get("If-None-Match"); got != `"v1"` {
+				t.Fatalf("second request If-None-Match = %q, want %q", got, `"v1"`)
+			}
+			if got := r.Header.Get("If-Modified-Since"); got != "" {
+				t.Fatalf("second request If-Modified-Since = %q, want empty when ETag is cached", got)
+			}
+			w.WriteHeader(http.StatusNotModified)
+		default:
+			t.Fatalf("unexpected call count %d", call)
+		}
+	}))
+	defer srv.Close()
+
+	validators := HTTPValidators{}
+	body, changed, next, err := FetchBodyIfChanged(context.Background(), srv.Client(), srv.URL, "test-agent", "application/json", true, validators)
+	if err != nil {
+		t.Fatalf("first FetchBodyIfChanged() error = %v", err)
+	}
+	if !changed {
+		t.Fatalf("first FetchBodyIfChanged() changed = false, want true")
+	}
+	if got := string(body); got != `{"ok":true}` {
+		t.Fatalf("first FetchBodyIfChanged() body = %q", got)
+	}
+	if got := next.ETag; got != `"v1"` {
+		t.Fatalf("cached ETag = %q, want %q", got, `"v1"`)
+	}
+	if got := next.LastModified; got != "Mon, 02 Jan 2006 15:04:05 GMT" {
+		t.Fatalf("cached Last-Modified = %q", got)
+	}
+
+	body, changed, next, err = FetchBodyIfChanged(context.Background(), srv.Client(), srv.URL, "test-agent", "application/json", true, next)
+	if err != nil {
+		t.Fatalf("second FetchBodyIfChanged() error = %v", err)
+	}
+	if changed {
+		t.Fatalf("second FetchBodyIfChanged() changed = true, want false")
+	}
+	if body != nil {
+		t.Fatalf("second FetchBodyIfChanged() body = %q, want nil", string(body))
+	}
+	if got := next.ETag; got != `"v1"` {
+		t.Fatalf("cached ETag after 304 = %q, want %q", got, `"v1"`)
+	}
+}
+
+func TestFetchBodyIfChangedFallsBackToIfModifiedSince(t *testing.T) {
+	t.Helper()
+
+	var call int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		call++
+		switch call {
+		case 1:
+			w.Header().Set("Last-Modified", "Tue, 03 Jan 2006 15:04:05 GMT")
+			_, _ = w.Write([]byte(`first`))
+		case 2:
+			if got := r.Header.Get("If-None-Match"); got != "" {
+				t.Fatalf("second request If-None-Match = %q, want empty", got)
+			}
+			if got := r.Header.Get("If-Modified-Since"); got != "Tue, 03 Jan 2006 15:04:05 GMT" {
+				t.Fatalf("second request If-Modified-Since = %q", got)
+			}
+			w.WriteHeader(http.StatusNotModified)
+		default:
+			t.Fatalf("unexpected call count %d", call)
+		}
+	}))
+	defer srv.Close()
+
+	_, changed, validators, err := FetchBodyIfChanged(context.Background(), srv.Client(), srv.URL, "test-agent", "", true, HTTPValidators{})
+	if err != nil {
+		t.Fatalf("first FetchBodyIfChanged() error = %v", err)
+	}
+	if !changed {
+		t.Fatalf("first FetchBodyIfChanged() changed = false, want true")
+	}
+	if got := validators.LastModified; got != "Tue, 03 Jan 2006 15:04:05 GMT" {
+		t.Fatalf("cached Last-Modified = %q", got)
+	}
+
+	_, changed, _, err = FetchBodyIfChanged(context.Background(), srv.Client(), srv.URL, "test-agent", "", true, validators)
+	if err != nil {
+		t.Fatalf("second FetchBodyIfChanged() error = %v", err)
+	}
+	if changed {
+		t.Fatalf("second FetchBodyIfChanged() changed = true, want false")
+	}
+}
+
+func TestFetchBodyIfChangedClearsValidatorsOn200WithoutValidators(t *testing.T) {
+	t.Helper()
+
+	var call int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		call++
+		switch call {
+		case 1:
+			w.Header().Set("ETag", `"v1"`)
+			_, _ = w.Write([]byte(`first`))
+		case 2:
+			if got := r.Header.Get("If-None-Match"); got != `"v1"` {
+				t.Fatalf("second request If-None-Match = %q", got)
+			}
+			_, _ = w.Write([]byte(`second`))
+		case 3:
+			if got := r.Header.Get("If-None-Match"); got != "" {
+				t.Fatalf("third request If-None-Match = %q, want empty", got)
+			}
+			if got := r.Header.Get("If-Modified-Since"); got != "" {
+				t.Fatalf("third request If-Modified-Since = %q, want empty", got)
+			}
+			_, _ = w.Write([]byte(`third`))
+		default:
+			t.Fatalf("unexpected call count %d", call)
+		}
+	}))
+	defer srv.Close()
+
+	_, _, validators, err := FetchBodyIfChanged(context.Background(), srv.Client(), srv.URL, "test-agent", "", true, HTTPValidators{})
+	if err != nil {
+		t.Fatalf("first FetchBodyIfChanged() error = %v", err)
+	}
+	_, _, validators, err = FetchBodyIfChanged(context.Background(), srv.Client(), srv.URL, "test-agent", "", true, validators)
+	if err != nil {
+		t.Fatalf("second FetchBodyIfChanged() error = %v", err)
+	}
+	if validators.ETag != "" || validators.LastModified != "" {
+		t.Fatalf("validators after 200 without validators = %+v, want cleared", validators)
+	}
+	_, _, _, err = FetchBodyIfChanged(context.Background(), srv.Client(), srv.URL, "test-agent", "", true, validators)
+	if err != nil {
+		t.Fatalf("third FetchBodyIfChanged() error = %v", err)
+	}
+}
+
+func TestFetchBodyIfChangedConditionalDisabledSkipsConditionalHeaders(t *testing.T) {
+	t.Helper()
+
+	var calls int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		if got := r.Header.Get("If-None-Match"); got != "" {
+			t.Fatalf("request If-None-Match = %q, want empty", got)
+		}
+		if got := r.Header.Get("If-Modified-Since"); got != "" {
+			t.Fatalf("request If-Modified-Since = %q, want empty", got)
+		}
+		_, _ = w.Write([]byte(`body`))
+	}))
+	defer srv.Close()
+
+	validators := HTTPValidators{ETag: `"v1"`, LastModified: "Wed, 04 Jan 2006 15:04:05 GMT"}
+	_, changed, next, err := FetchBodyIfChanged(context.Background(), srv.Client(), srv.URL, "test-agent", "", false, validators)
+	if err != nil {
+		t.Fatalf("FetchBodyIfChanged() error = %v", err)
+	}
+	if !changed {
+		t.Fatalf("FetchBodyIfChanged() changed = false, want true")
+	}
+	if next != validators {
+		t.Fatalf("validators changed when conditional disabled: got %+v want %+v", next, validators)
+	}
+	if calls != 1 {
+		t.Fatalf("calls = %d, want 1", calls)
+	}
+}
+
+func TestFetchBodyIfChangedAllowsEmpty304ButRejectsEmpty200(t *testing.T) {
+	t.Helper()
+
+	notModified := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusNotModified)
+	}))
+	defer notModified.Close()
+
+	_, changed, _, err := FetchBodyIfChanged(
+		context.Background(),
+		notModified.Client(),
+		notModified.URL,
+		"test-agent",
+		"",
+		true,
+		HTTPValidators{ETag: `"v1"`},
+	)
+	if err != nil {
+		t.Fatalf("304 FetchBodyIfChanged() error = %v", err)
+	}
+	if changed {
+		t.Fatalf("304 FetchBodyIfChanged() changed = true, want false")
+	}
+
+	emptyBody := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer emptyBody.Close()
+
+	_, _, _, err = FetchBodyIfChanged(context.Background(), emptyBody.Client(), emptyBody.URL, "test-agent", "", true, HTTPValidators{})
+	if err == nil {
+		t.Fatalf("empty 200 FetchBodyIfChanged() error = nil, want error")
+	}
+	if err.Error() != "empty response body" {
+		t.Fatalf("empty 200 FetchBodyIfChanged() error = %q", err)
+	}
+}