Moved common HTTP polling helpers into feedkit and implemented support for ETag and Last-Modified

- add new `processors` package with canonical `Processor` interface
- add `processors.Registry` with Register/Build/BuildChain factory model
- switch `pipeline.Pipeline` to `[]processors.Processor`
- replace `normalize.Registry` + registry adapter with direct `normalize.Processor`
- remove `normalize/registry.go`
- update root docs to position normalize as one optional processing stage
- add tests for processors registry, normalize processor behavior, and pipeline flow

BREAKING CHANGE:
- `pipeline.Processor` removed; use `processors.Processor`
- `normalize.Registry` and old normalize processor adapter APIs removed
- downstream daemons must update processor wiring to new `processors.Registry`
  and `normalize.NewProcessor(...)`

README.md

@@ -15,13 +15,12 @@ lifecycle, domain schemas, and domain-specific validation in your daemon.
 
 ## Conceptual pipeline
 
-Collect -> Normalize (optional) -> Policy -> Route -> Emit
+Collect -> Process (optional stages, including dedupe + normalize) -> Route -> Emit
 
 | Stage | Package(s) |
 |---|---|
 | Collect | `sources`, `scheduler` |
-| Normalize | `normalize` (optional in `pipeline`) |
-| Policy | `pipeline` |
+| Process | `pipeline`, `processors`, `processors/dedupe`, `processors/normalize` (optional stages) |
 | Route | `dispatch` |
 | Emit | `sinks` |
 | Configure | `config` |
@@ -82,10 +81,19 @@ Runs one goroutine per source job:
 Optional processing chain between collection and dispatch.
 Processors can transform, drop, or reject events.
 
-### `normalize`
+### `processors`
 
-Optional normalization package (already implemented). Typical use: sources emit raw
-payload events, then normalize to canonical schemas in a pipeline stage.
+Defines the generic processor interface and a named-driver registry used by
+daemons to build ordered processor chains.
+
+### `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.
 
 ### `dispatch`
 
@@ -93,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
-additional sink implementations at varying maturity.
+Defines sink interface and sink registry. Built-ins include:
+- `stdout`
+- `nats`
+- `postgres`
+
+Detailed Postgres configuration and wiring examples live in package docs:
+`sinks/doc.go`.
 
 ## Typical wiring
 

doc.go

@@ -5,13 +5,12 @@
 //
 // Conceptual flow:
 //
-//	Collect -> Normalize (optional) -> Policy -> Route -> Emit
+//	Collect -> Process (optional stages, including dedupe + normalize) -> Route -> Emit
 //
 // In feedkit this maps to:
 //
 //	Collect: sources + scheduler
-//	Normalize: normalize (optional pipeline stage)
-//	Policy:    pipeline
+//	Process: pipeline + processors + processors/dedupe + processors/normalize (optional stages)
 //	Route:   dispatch
 //	Emit:    sinks
 //	Config:  config
@@ -51,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:
 //
@@ -62,8 +74,14 @@
 //     Processor chain between scheduler and dispatch.
 //     Processors can transform, drop, or reject events.
 //
-//   - normalize
-//     Optional pipeline processor for raw->canonical mapping.
+//   - processors
+//     Generic processor interface and named factory registry for wiring chains.
+//
+//   - 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.
 //
 //   - dispatch
@@ -71,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=

normalize/registry.go

@@ -1,140 +0,0 @@
-package normalize
-
-import (
-	"context"
-	"fmt"
-	"sync"
-
-	"gitea.maximumdirect.net/ejr/feedkit/event"
-)
-
-// Registry holds a set of Normalizers and selects one for a given event.
-//
-// Selection rule (simple + predictable):
-//   - iterate in registration order
-//   - the FIRST Normalizer whose Match(e) returns true is used
-//
-// If none match, the event passes through unchanged.
-//
-// Why "first match wins"?
-// Normalization is usually a single mapping step from a raw schema/version into
-// a normalized schema/version. If you want multiple transformation steps,
-// model them as multiple pipeline processors (which feedkit already supports).
-type Registry struct {
-	mu sync.RWMutex
-	ns []Normalizer
-}
-
-// Register adds a normalizer to the registry.
-//
-// Register panics if n is nil; this is a programmer error and should fail fast.
-func (r *Registry) Register(n Normalizer) {
-	if n == nil {
-		panic("normalize.Registry.Register: normalizer cannot be nil")
-	}
-	r.mu.Lock()
-	defer r.mu.Unlock()
-	r.ns = append(r.ns, n)
-}
-
-// Normalize finds the first matching Normalizer and applies it.
-//
-// If no normalizer matches, it returns the input event unchanged.
-//
-// If a normalizer returns (nil, nil), the event is dropped.
-func (r *Registry) Normalize(ctx context.Context, in event.Event) (*event.Event, error) {
-	if r == nil {
-		// Nil registry is a valid "feature off" state.
-		out := in
-		return &out, nil
-	}
-
-	r.mu.RLock()
-	ns := append([]Normalizer(nil), r.ns...) // copy for safe iteration outside lock
-	r.mu.RUnlock()
-
-	for _, n := range ns {
-		if n == nil {
-			// Shouldn't happen (Register panics), but guard anyway.
-			continue
-		}
-		if !n.Match(in) {
-			continue
-		}
-
-		out, err := n.Normalize(ctx, in)
-		if err != nil {
-			return nil, fmt.Errorf("normalize: normalizer failed: %w", err)
-		}
-		// out may be nil to signal "drop".
-		return out, nil
-	}
-
-	// No match: pass through unchanged.
-	out := in
-	return &out, nil
-}
-
-// Processor adapts a Registry into a pipeline Processor.
-//
-// It implements:
-//
-//	Process(ctx context.Context, in event.Event) (*event.Event, error)
-//
-// which matches feedkit/pipeline.Processor.
-//
-// Optionality:
-//   - If Registry is nil, Processor becomes a no-op pass-through.
-//   - If Registry has no matching normalizer for an event, that event passes through unchanged.
-type Processor struct {
-	Registry *Registry
-
-	// If true, events that do not match any normalizer cause an error.
-	// Default is false (pass-through), which is the behavior you asked for.
-	RequireMatch bool
-}
-
-// Process implements the pipeline.Processor interface.
-func (p Processor) Process(ctx context.Context, in event.Event) (*event.Event, error) {
-	// "Feature off": no registry means no normalization.
-	if p.Registry == nil {
-		out := in
-		return &out, nil
-	}
-
-	out, err := p.Registry.Normalize(ctx, in)
-	if err != nil {
-		return nil, err
-	}
-
-	if out == nil {
-		// Dropped by normalization policy.
-		return nil, nil
-	}
-
-	if p.RequireMatch {
-		// Detect "no-op pass-through due to no match" by checking whether a match existed.
-		// We do this with a cheap second pass to avoid changing Normalize()'s signature.
-		// (This is rare to enable; correctness/clarity > micro-optimization.)
-		if !p.Registry.hasMatch(in) {
-			return nil, fmt.Errorf("normalize: no normalizer matched event (id=%s kind=%s source=%s schema=%q)",
-				in.ID, in.Kind, in.Source, in.Schema)
-		}
-	}
-
-	return out, nil
-}
-
-func (r *Registry) hasMatch(in event.Event) bool {
-	if r == nil {
-		return false
-	}
-	r.mu.RLock()
-	defer r.mu.RUnlock()
-	for _, n := range r.ns {
-		if n != nil && n.Match(in) {
-			return true
-		}
-	}
-	return false
-}

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

pipeline/pipeline.go

@@ -5,15 +5,11 @@ import (
 	"fmt"
 
 	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/feedkit/processors"
 )
 
-// Processor can mutate/drop events (dedupe, rate-limit, normalization tweaks).
-type Processor interface {
-	Process(ctx context.Context, in event.Event) (out *event.Event, err error)
-}
-
 type Pipeline struct {
-	Processors []Processor
+	Processors []processors.Processor
 }
 
 func (p *Pipeline) Process(ctx context.Context, e event.Event) (*event.Event, error) {

pipeline/pipeline_test.go

@@ -0,0 +1,115 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+	"gitea.maximumdirect.net/ejr/feedkit/processors"
+)
+
+type procFunc func(context.Context, event.Event) (*event.Event, error)
+
+func (f procFunc) Process(ctx context.Context, in event.Event) (*event.Event, error) {
+	return f(ctx, in)
+}
+
+func TestPipelineProcessSequentialOrder(t *testing.T) {
+	var gotOrder []string
+
+	p := &Pipeline{
+		Processors: []processors.Processor{
+			procFunc(func(_ context.Context, in event.Event) (*event.Event, error) {
+				gotOrder = append(gotOrder, "first")
+				out := in
+				out.Schema = "stage.one.v1"
+				return &out, nil
+			}),
+			procFunc(func(_ context.Context, in event.Event) (*event.Event, error) {
+				gotOrder = append(gotOrder, "second")
+				if in.Schema != "stage.one.v1" {
+					return nil, fmt.Errorf("expected schema from first stage, got %q", in.Schema)
+				}
+				out := in
+				out.Schema = "stage.two.v1"
+				return &out, nil
+			}),
+		},
+	}
+
+	out, err := p.Process(context.Background(), validEvent())
+	if err != nil {
+		t.Fatalf("Process error: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("expected output event, got nil")
+	}
+	if out.Schema != "stage.two.v1" {
+		t.Fatalf("unexpected output schema: %q", out.Schema)
+	}
+	if strings.Join(gotOrder, ",") != "first,second" {
+		t.Fatalf("unexpected processor order: %v", gotOrder)
+	}
+}
+
+func TestPipelineProcessInvalidInput(t *testing.T) {
+	p := &Pipeline{}
+	_, err := p.Process(context.Background(), event.Event{})
+	if err == nil {
+		t.Fatalf("expected input validation error")
+	}
+	if !strings.Contains(err.Error(), "invalid input event") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestPipelineProcessDrop(t *testing.T) {
+	p := &Pipeline{
+		Processors: []processors.Processor{
+			procFunc(func(context.Context, event.Event) (*event.Event, error) {
+				return nil, nil
+			}),
+		},
+	}
+
+	out, err := p.Process(context.Background(), validEvent())
+	if err != nil {
+		t.Fatalf("Process error: %v", err)
+	}
+	if out != nil {
+		t.Fatalf("expected nil output for dropped event, got %#v", out)
+	}
+}
+
+func TestPipelineProcessInvalidOutput(t *testing.T) {
+	p := &Pipeline{
+		Processors: []processors.Processor{
+			procFunc(func(_ context.Context, in event.Event) (*event.Event, error) {
+				out := in
+				out.Payload = nil
+				return &out, nil
+			}),
+		},
+	}
+
+	_, err := p.Process(context.Background(), validEvent())
+	if err == nil {
+		t.Fatalf("expected output validation error")
+	}
+	if !strings.Contains(err.Error(), "invalid output event") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func validEvent() event.Event {
+	return event.Event{
+		ID:        "evt-1",
+		Kind:      event.Kind("observation"),
+		Source:    "source-1",
+		EmittedAt: time.Now().UTC(),
+		Payload:   map[string]any{"ok": true},
+	}
+}

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

@@ -0,0 +1,24 @@
+// Package processors defines feedkit's generic processor abstraction and registry.
+//
+// Processors are optional pipeline stages that can transform, drop, or reject
+// events before dispatch to sinks.
+//
+// Registry provides name-based construction so daemons can assemble processor
+// chains without embedding switch statements in wiring code.
+//
+// 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{"dedupe", "normalize"})
+//	if err != nil {
+//	    // handle wiring error
+//	}
+//
+//	p := &pipeline.Pipeline{Processors: chain}
+package processors

processors/normalize/doc.go

@@ -1,4 +1,4 @@
-// Package normalize provides an OPTIONAL normalization hook for feedkit pipelines.
+// Package normalize provides a concrete normalization processor for feedkit pipelines.
 //
 // Motivation:
 // Many daemons have sources that:
@@ -9,9 +9,9 @@
 // encourages duplication (unit conversions, common mapping helpers, etc.).
 //
 // This package lets a source emit a "raw" event (e.g., Schema="raw.openweather.current.v1",
-// Payload=json.RawMessage), and then a normalization processor can convert it into a
+// Payload=json.RawMessage), and then a normalize.Processor can convert it into a
 // normalized event (e.g., Schema="weather.observation.v1", Payload=WeatherObservation{}).
 //
 // Key property: normalization is optional.
-// If no registered Normalizer matches an event, it passes through unchanged.
+// If no Normalizer matches an event, Processor passes it through unchanged by default.
 package normalize

processors/normalize/processor.go

@@ -0,0 +1,57 @@
+package normalize
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+// Processor applies ordered normalization rules to pipeline events.
+//
+// Selection rule:
+//   - iterate in Normalizers order
+//   - the first Normalizer whose Match returns true is applied
+//
+// If no normalizer matches, the default behavior is pass-through.
+type Processor struct {
+	Normalizers []Normalizer
+
+	// If true, events that do not match any normalizer cause an error.
+	// Default is false (pass-through).
+	RequireMatch bool
+}
+
+// NewProcessor constructs a normalization processor from an ordered normalizer list.
+func NewProcessor(normalizers []Normalizer, requireMatch bool) Processor {
+	return Processor{
+		Normalizers:  append([]Normalizer(nil), normalizers...),
+		RequireMatch: requireMatch,
+	}
+}
+
+// Process implements processors.Processor.
+func (p Processor) Process(ctx context.Context, in event.Event) (*event.Event, error) {
+	for _, n := range p.Normalizers {
+		if n == nil {
+			continue
+		}
+		if !n.Match(in) {
+			continue
+		}
+
+		out, err := n.Normalize(ctx, in)
+		if err != nil {
+			return nil, fmt.Errorf("normalize: normalizer failed: %w", err)
+		}
+		return out, nil
+	}
+
+	if p.RequireMatch {
+		return nil, fmt.Errorf("normalize: no normalizer matched event (id=%s kind=%s source=%s schema=%q)",
+			in.ID, in.Kind, in.Source, in.Schema)
+	}
+
+	out := in
+	return &out, nil
+}

processors/normalize/processor_test.go

@@ -0,0 +1,139 @@
+package normalize
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+func TestProcessorFirstMatchWins(t *testing.T) {
+	var firstCalls, secondCalls int
+
+	p := NewProcessor([]Normalizer{
+		Func{
+			MatchFn: func(event.Event) bool { return true },
+			NormalizeFn: func(_ context.Context, in event.Event) (*event.Event, error) {
+				firstCalls++
+				out := in
+				out.Schema = "normalized.first.v1"
+				return &out, nil
+			},
+		},
+		Func{
+			MatchFn: func(event.Event) bool { return true },
+			NormalizeFn: func(_ context.Context, in event.Event) (*event.Event, error) {
+				secondCalls++
+				out := in
+				out.Schema = "normalized.second.v1"
+				return &out, nil
+			},
+		},
+	}, false)
+
+	out, err := p.Process(context.Background(), testEvent())
+	if err != nil {
+		t.Fatalf("Process error: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("expected output event, got nil")
+	}
+	if out.Schema != "normalized.first.v1" {
+		t.Fatalf("unexpected schema: %q", out.Schema)
+	}
+	if firstCalls != 1 {
+		t.Fatalf("expected first normalizer called once, got %d", firstCalls)
+	}
+	if secondCalls != 0 {
+		t.Fatalf("expected second normalizer skipped, got %d calls", secondCalls)
+	}
+}
+
+func TestProcessorNoMatchPassThroughAndRequireMatch(t *testing.T) {
+	in := testEvent()
+	in.Schema = "raw.schema.v1"
+
+	passThrough := NewProcessor([]Normalizer{
+		Func{
+			MatchFn: func(event.Event) bool { return false },
+			NormalizeFn: func(_ context.Context, in event.Event) (*event.Event, error) {
+				out := in
+				out.Schema = "should.not.run"
+				return &out, nil
+			},
+		},
+	}, false)
+
+	out, err := passThrough.Process(context.Background(), in)
+	if err != nil {
+		t.Fatalf("pass-through Process error: %v", err)
+	}
+	if out == nil {
+		t.Fatalf("expected pass-through output event, got nil")
+	}
+	if out.Schema != "raw.schema.v1" {
+		t.Fatalf("expected unchanged schema, got %q", out.Schema)
+	}
+
+	required := NewProcessor(nil, true)
+	_, err = required.Process(context.Background(), in)
+	if err == nil {
+		t.Fatalf("expected require-match error")
+	}
+	if !strings.Contains(err.Error(), "no normalizer matched") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestProcessorDropAndErrorPropagation(t *testing.T) {
+	t.Run("drop", func(t *testing.T) {
+		p := NewProcessor([]Normalizer{
+			Func{
+				MatchFn: func(event.Event) bool { return true },
+				NormalizeFn: func(context.Context, event.Event) (*event.Event, error) {
+					return nil, nil
+				},
+			},
+		}, false)
+
+		out, err := p.Process(context.Background(), testEvent())
+		if err != nil {
+			t.Fatalf("Process error: %v", err)
+		}
+		if out != nil {
+			t.Fatalf("expected nil output for dropped event, got %#v", out)
+		}
+	})
+
+	t.Run("error", func(t *testing.T) {
+		p := NewProcessor([]Normalizer{
+			Func{
+				MatchFn: func(event.Event) bool { return true },
+				NormalizeFn: func(context.Context, event.Event) (*event.Event, error) {
+					return nil, errors.New("map failed")
+				},
+			},
+		}, false)
+
+		_, err := p.Process(context.Background(), testEvent())
+		if err == nil {
+			t.Fatalf("expected error")
+		}
+		if !strings.Contains(err.Error(), "normalizer failed") {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
+
+func testEvent() event.Event {
+	return event.Event{
+		ID:        "evt-normalize-1",
+		Kind:      event.Kind("observation"),
+		Source:    "source-1",
+		EmittedAt: time.Now().UTC(),
+		Payload:   map[string]any{"x": 1},
+	}
+}

processors/processor.go

@@ -0,0 +1,15 @@
+package processors
+
+import (
+	"context"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+// Processor can mutate/drop events (dedupe, rate-limit, normalization tweaks).
+type Processor interface {
+	Process(ctx context.Context, in event.Event) (out *event.Event, err error)
+}
+
+// Factory constructs a configured Processor instance.
+type Factory func() (Processor, error)

processors/registry.go

@@ -0,0 +1,71 @@
+package processors
+
+import (
+	"fmt"
+	"strings"
+)
+
+type Registry struct {
+	byDriver map[string]Factory
+}
+
+func NewRegistry() *Registry {
+	return &Registry{byDriver: map[string]Factory{}}
+}
+
+// Register associates a processor driver name with a factory.
+//
+// Register panics for empty driver names, nil factories, and duplicates.
+func (r *Registry) Register(driver string, f Factory) {
+	if r == nil {
+		panic("processors.Registry.Register: registry cannot be nil")
+	}
+	driver = strings.TrimSpace(driver)
+	if driver == "" {
+		panic("processors.Registry.Register: driver cannot be empty")
+	}
+	if f == nil {
+		panic(fmt.Sprintf("processors.Registry.Register: factory cannot be nil (driver=%q)", driver))
+	}
+	if r.byDriver == nil {
+		r.byDriver = map[string]Factory{}
+	}
+	if _, exists := r.byDriver[driver]; exists {
+		panic(fmt.Sprintf("processors.Registry.Register: driver %q already registered", driver))
+	}
+	r.byDriver[driver] = f
+}
+
+// Build constructs a Processor by driver name.
+func (r *Registry) Build(driver string) (Processor, error) {
+	if r == nil {
+		return nil, fmt.Errorf("processors registry is nil")
+	}
+	driver = strings.TrimSpace(driver)
+	f, ok := r.byDriver[driver]
+	if !ok {
+		return nil, fmt.Errorf("unknown processor driver: %q", driver)
+	}
+
+	p, err := f()
+	if err != nil {
+		return nil, fmt.Errorf("build processor %q: %w", driver, err)
+	}
+	if p == nil {
+		return nil, fmt.Errorf("build processor %q: factory returned nil processor", driver)
+	}
+	return p, nil
+}
+
+// BuildChain constructs an ordered processor chain from a driver list.
+func (r *Registry) BuildChain(drivers []string) ([]Processor, error) {
+	out := make([]Processor, 0, len(drivers))
+	for i, driver := range drivers {
+		p, err := r.Build(driver)
+		if err != nil {
+			return nil, fmt.Errorf("build processor chain[%d] (%q): %w", i, strings.TrimSpace(driver), err)
+		}
+		out = append(out, p)
+	}
+	return out, nil
+}

processors/registry_test.go

@@ -0,0 +1,100 @@
+package processors
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/ejr/feedkit/event"
+)
+
+type testProcessor struct {
+	name string
+}
+
+func (p testProcessor) Process(context.Context, event.Event) (*event.Event, error) {
+	return nil, nil
+}
+
+func TestRegistryRegisterValidation(t *testing.T) {
+	t.Run("empty driver panics", func(t *testing.T) {
+		r := NewRegistry()
+		assertPanics(t, func() {
+			r.Register("   ", func() (Processor, error) { return testProcessor{name: "x"}, nil })
+		})
+	})
+
+	t.Run("nil factory panics", func(t *testing.T) {
+		r := NewRegistry()
+		assertPanics(t, func() {
+			r.Register("normalize", nil)
+		})
+	})
+
+	t.Run("duplicate driver panics", func(t *testing.T) {
+		r := NewRegistry()
+		r.Register("normalize", func() (Processor, error) { return testProcessor{name: "a"}, nil })
+		assertPanics(t, func() {
+			r.Register("normalize", func() (Processor, error) { return testProcessor{name: "b"}, nil })
+		})
+	})
+}
+
+func TestRegistryBuildUnknownDriver(t *testing.T) {
+	r := NewRegistry()
+	_, err := r.Build("does_not_exist")
+	if err == nil {
+		t.Fatalf("expected error for unknown driver")
+	}
+	if !strings.Contains(err.Error(), "unknown processor driver") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestRegistryBuildChainPreservesOrder(t *testing.T) {
+	r := NewRegistry()
+	r.Register("first", func() (Processor, error) { return testProcessor{name: "first"}, nil })
+	r.Register("second", func() (Processor, error) { return testProcessor{name: "second"}, nil })
+
+	chain, err := r.BuildChain([]string{"first", "second"})
+	if err != nil {
+		t.Fatalf("BuildChain error: %v", err)
+	}
+	if len(chain) != 2 {
+		t.Fatalf("expected 2 processors, got %d", len(chain))
+	}
+
+	p0, ok := chain[0].(testProcessor)
+	if !ok || p0.name != "first" {
+		t.Fatalf("unexpected chain[0]: %#v", chain[0])
+	}
+	p1, ok := chain[1].(testProcessor)
+	if !ok || p1.name != "second" {
+		t.Fatalf("unexpected chain[1]: %#v", chain[1])
+	}
+}
+
+func TestRegistryBuildChainIndexedFailure(t *testing.T) {
+	r := NewRegistry()
+	r.Register("ok", func() (Processor, error) { return testProcessor{name: "ok"}, nil })
+	r.Register("broken", func() (Processor, error) { return nil, errors.New("boom") })
+
+	_, err := r.BuildChain([]string{"ok", "broken"})
+	if err == nil {
+		t.Fatalf("expected error")
+	}
+	if !strings.Contains(err.Error(), "chain[1]") {
+		t.Fatalf("expected indexed error, got: %v", err)
+	}
+}
+
+func assertPanics(t *testing.T, fn func()) {
+	t.Helper()
+	defer func() {
+		if recover() == nil {
+			t.Fatalf("expected panic")
+		}
+	}()
+	fn()
+}

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/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 {
-	name string
-	dsn  string
+const postgresInitTimeout = 5 * time.Second
+
+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) {
-	dsn, err := requireStringParam(cfg, "dsn")
+type postgresExecer interface {
+	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/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)
 	}
-
-	res, err := client.Do(req)
-	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)
+	if headerName != "" && headerValue != "" {
+		req.Header.Set(headerName, headerValue)
 	}
 
+	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)
+	}
+}