ejr/feedkit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Cleaned up documentation and removed stubs and TODOs throughout the application

Browse Source
This commit is contained in:
Eric Rakestraw
2026-03-28 13:02:37 -05:00
parent 3ef93faf69
commit 3281368922
18 changed files with 403 additions and 345 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
README.md
View File

@@ -1,127 +1,92 @@
# feedkit # feedkit
`feedkit` provides domain-agnostic plumbing for feed-processing daemons. `feedkit` is a small Go toolkit for building feed-processing daemons.
A daemon built on feedkit typically: It gives you the reusable plumbing around collection, processing, routing, and
- ingests upstream input (polling APIs or consuming streams) emission, while leaving domain concepts, schemas, and application wiring in
your daemon. The intended shape is a family of sibling applications such as
`weatherfeeder`, `newsfeeder`, or `earthquakefeeder` that all share the same
infrastructure patterns without sharing domain logic.
## What It Does
A daemon built on `feedkit` typically:
- ingests upstream input by polling HTTP APIs or consuming streams
- emits domain-agnostic `event.Event` values - emits domain-agnostic `event.Event` values
- applies optional processing (normalization, dedupe, policy) - optionally processes those events with stages like dedupe or normalization
- routes events to sinks (stdout, NATS, files, databases, etc.) - routes events to one or more sinks such as stdout, NATS, or Postgres
Conceptually, the pipeline is:
`Collect -> Process -> Route -> Emit`
## Philosophy ## Philosophy
feedkit is not a framework. It provides small composable packages and leaves `feedkit` is intentionally not a framework.
lifecycle, domain schemas, and domain-specific validation in your daemon.
## Conceptual pipeline It does not try to own:
- your domain payload schemas
- your domain event kinds
- your daemon lifecycle or `main.go`
- your observability stack or deployment model
Collect -> Process (optional stages, including dedupe + normalize) -> Route -> Emit Instead, it provides small composable packages that are easy to wire together in
different daemons.
| Stage | Package(s) | ## When To Use It
|---|---|
| Collect | `sources`, `scheduler` |
| Process | `pipeline`, `processors`, `processors/dedupe`, `processors/normalize` (optional stages) |
| Route | `dispatch` |
| Emit | `sinks` |
| Configure | `config` |
## Core packages `feedkit` is a good fit when you want:
- multiple small ingestion daemons with shared infrastructure patterns
- clear separation between raw upstream payloads and normalized canonical models
- reusable routing and sink behavior across domains
- strong config and event-envelope conventions without centralizing domain rules
### `config` It is a poor fit if you want a monolithic framework that dictates application
structure end-to-end.
Loads YAML config with strict decoding and domain-agnostic validation. ## Built-In Capabilities
`SourceConfig` supports both source modes: `feedkit` currently includes:
- `mode: poll` requires `every` - strict YAML config loading and validation
- `mode: stream` forbids `every` - polling and streaming source abstractions
- omitted `mode` means auto (inferred from the registered driver type) - scheduler orchestration for configured sources
- optional pipeline processors
- built-in dedupe and normalization processors
- route compilation and sink fanout
- built-in sinks for `stdout`, `nats`, and `postgres`
It also supports optional expected source kinds: The Postgres sink is intentionally split between feedkit-owned infrastructure
- `kinds: ["observation", "alert"]` (preferred) and daemon-owned schema mapping. `feedkit` manages connection setup, DDL,
- `kind: "observation"` (legacy fallback) writes, and pruning; downstream applications define the schema and event mapper.
### `event` ## Typical Wiring
Defines the domain-agnostic event envelope (`event.Event`) used across the system. At a high level, a daemon built on `feedkit` does this:
### `sources`
Defines source interfaces and driver registry:
```go
type Input interface {
Name() string
}
type PollSource interface {
Input
Poll(ctx context.Context) ([]event.Event, error)
}
type StreamSource interface {
Input
Run(ctx context.Context, out chan<- event.Event) error
}
```
Notes:
- a poll can emit `0..N` events
- stream sources emit events continuously
- a single source may emit multiple event kinds
- driver implementations live in downstream daemons and are registered via `sources.Registry`
### `scheduler`
Runs one goroutine per source job:
- poll sources: cadence driven (`every` + jitter)
- stream sources: continuous run loop
### `pipeline`
Optional processing chain between collection and dispatch.
Processors can transform, drop, or reject events.
### `processors`
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`
Compiles routes and fans out events to sinks with per-sink queue/worker isolation.
### `sinks`
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
1. Load config. 1. Load config.
2. Register/build sources from `cfg.Sources`. 2. Register domain-specific source drivers.
3. Register/build sinks from `cfg.Sinks`. 3. Register built-in and/or custom sinks.
4. Compile routes. 4. Build sources, sinks, and optional processor chain from config.
5. Start scheduler (`sources -> bus`). 5. Compile routes.
6. Start dispatcher (`bus -> pipeline -> sinks`). 6. Start the scheduler and dispatcher.
## Non-goals The package docs are the better source of truth for code-level details. In
particular, each subpackage `doc.go` describes its external API surface and any
optional helper APIs in `helpers.go`.
feedkit intentionally does not: ## Package Layout
- define domain payload schemas
- enforce domain-specific event kinds The major packages are:
- own application lifecycle - `config`: config loading and validation
- prescribe observability stack choices - `event`: the domain-agnostic event envelope
- `sources`: source interfaces and reusable source helpers
- `scheduler`: source execution and cadence management
- `processors`: processor interfaces and registry
- `processors/dedupe`: built-in in-memory dedupe processor
- `processors/normalize`: built-in normalization processor and helpers
- `pipeline`: optional processor chain
- `dispatch`: route compilation and fanout
- `sinks`: sink interfaces, built-ins, and Postgres registration helpers
The root package docs in `doc.go` provide a concise package-by-package map for
Go documentation consumers.

2
config/config.go
View File

@@ -101,7 +101,7 @@ func (cfg SourceConfig) ExpectedKinds() []string {
// SinkConfig describes one output sink adapter. // SinkConfig describes one output sink adapter.
type SinkConfig struct { type SinkConfig struct {
Name string `yaml:"name"` Name string `yaml:"name"`
Driver string `yaml:"driver"` // "stdout", "file", "postgres", "rabbitmq", ... Driver string `yaml:"driver"` // "stdout", "nats", "postgres", ...
Params map[string]any `yaml:"params"` // sink-specific settings Params map[string]any `yaml:"params"` // sink-specific settings
} }

2
config/load.go
View File

@@ -141,7 +141,7 @@ func (c *Config) Validate() error {
} }
if strings.TrimSpace(s.Driver) == "" { if strings.TrimSpace(s.Driver) == "" {
m.Add(fieldErr(path+".driver", "is required (stdout|file|postgres|rabbitmq|...)")) m.Add(fieldErr(path+".driver", "is required (stdout|nats|postgres|...)"))
} }
// Params can be nil; that's fine. // Params can be nil; that's fine.

136
doc.go
View File

@@ -1,130 +1,56 @@
// Package feedkit provides domain-agnostic plumbing for feed-processing daemons. // Package feedkit provides a high-level map of the feedkit package set.
// //
// A feed daemon ingests upstream input, turns it into event.Event values, applies // Most real applications do not import the root package directly. Instead, they
// optional processing, and emits to sinks. // compose the subpackages that handle configuration, collection, processing,
// routing, and sinks.
// //
// Conceptual flow: // The usual flow through feedkit is:
// //
// Collect -> Process (optional stages, including dedupe + normalize) -> Route -> Emit // Collect -> Process -> Route -> Emit
// //
// In feedkit this maps to: // That flow maps to packages like this:
//
// Collect: sources + scheduler
// Process: pipeline + processors + processors/dedupe + processors/normalize (optional stages)
// Route: dispatch
// Emit: sinks
// Config: config
//
// feedkit intentionally does not define domain payload schemas or domain-specific
// validation rules. Those belong in each concrete daemon.
//
// Public packages
// //
// - config // - config
// YAML config loading/validation (strict decode + domain-agnostic checks). // Loads and validates daemon config. This package owns domain-agnostic
// // config shape and consistency checks.
// SourceConfig supports both polling and streaming sources:
//
// - mode: "poll" | "stream" | omitted (auto by driver type)
//
// - every: poll interval (required for mode="poll")
//
// - kinds: optional expected emitted kinds
//
// - kind: legacy singular fallback
//
// - params: driver-specific settings
// //
// - event // - event
// Domain-agnostic event envelope (ID, Kind, Source, EmittedAt, Schema, Payload). // Defines the event.Event envelope shared across sources, processors,
// dispatch, and sinks.
// //
// - sources // - sources
// Source abstractions and source-driver registry. // Defines polling and streaming source interfaces, the source registry, and
// // reusable source helpers.
// There are two source interfaces:
//
// - PollSource: Poll(ctx) ([]event.Event, error)
//
// - StreamSource: Run(ctx, out) error
//
// 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 // - scheduler
// Runs one goroutine per job: // Runs configured sources on a cadence or as long-lived stream workers.
//
// - PollSource jobs run on Every (+ jitter)
//
// - StreamSource jobs run continuously
//
// - pipeline
// Processor chain between scheduler and dispatch.
// Processors can transform, drop, or reject events.
// //
// - processors // - processors
// Generic processor interface and named factory registry for wiring chains. // Defines the generic processor interface and registry used to build
// ordered processor chains.
// //
// - processors/dedupe // - processors/dedupe
// Built-in in-memory LRU dedupe processor keyed by Event.ID. // Built-in in-memory dedupe processor keyed by Event.ID.
// //
// - processors/normalize // - processors/normalize
// Concrete pipeline processor for raw->canonical mapping. // Built-in normalization processor plus helper APIs for raw-to-canonical
// If no normalizer matches, the event passes through unchanged by default. // event mapping.
//
// - pipeline
// Applies an ordered processor chain between collection and dispatch.
// //
// - dispatch // - dispatch
// Routes events to sinks and isolates slow sinks via per-sink queues/workers. // Compiles routes and fans events out to sinks with per-sink isolation.
// //
// - sinks // - sinks
// Sink abstractions + sink registry. // Defines sink interfaces, the sink registry, built-in sinks, and Postgres
// Built-ins include stdout, NATS, and Postgres. For Postgres, downstream // schema registration helpers.
// 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) // feedkit is intentionally domain-agnostic. Domain schemas, domain event kinds,
// upstream-specific parsing, and daemon lifecycle remain the responsibility of
// each concrete application.
// //
// 1. Load config. // For repository-level overview and usage narrative, see README.md. For
// 2. Register source drivers and build sources from config.Sources. // code-level details, each subpackage doc.go is the source of truth for that
// 3. Register sink drivers and build sinks from config.Sinks. // package's public API surface and optional helpers.
// 4. Compile routes.
// 5. Start scheduler (sources -> bus) and dispatcher (bus -> pipeline -> sinks).
//
// Sketch:
//
// cfg, _ := config.Load("config.yml")
// srcReg := sources.NewRegistry()
// // domain registers poll/stream drivers...
//
// var jobs []scheduler.Job
// for _, sc := range cfg.Sources {
// src, _ := srcReg.BuildInput(sc)
// jobs = append(jobs, scheduler.Job{
// Source: src,
// Every: sc.Every.Duration,
// })
// }
//
// bus := make(chan event.Event, 256)
// s := &scheduler.Scheduler{Jobs: jobs, Out: bus, Logf: logf}
// // start dispatcher similarly...
//
// # Context and cancellation
//
// All blocking work should honor context cancellation:
// - source polling/streaming I/O
// - sink consumption
// - any expensive processor work
package feedkit package feedkit

5
pipeline/ratelimit.go
View File

@@ -1,5 +0,0 @@
package pipeline
// Placeholder for rate limit processor:
// - per source/kind sink routing limits
// - cooldown windows

22
processors/normalize/doc.go
View File

@@ -1,16 +1,18 @@
// Package normalize provides a concrete normalization processor for feedkit pipelines. // Package normalize provides the feedkit normalization processor and related
// helper APIs for raw-to-canonical event mapping.
// //
// Motivation: // External API surface:
// Many daemons have sources that: // - Processor: concrete processors.Processor implementation
// 1. fetch raw upstream data (often JSON), and // - Normalizer / Func: normalization interface and ergonomic function adapter
// 2. transform it into a domain's normalized payload format.
// //
// Doing both steps inside Source.Poll works, but tends to make sources large and // Optional helpers from helpers.go:
// encourages duplication (unit conversions, common mapping helpers, etc.). // - PayloadJSONBytes: extract supported JSON-shaped payloads into bytes
// - DecodeJSONPayload: decode an event payload into a typed struct
// - FinalizeEvent: copy the input event envelope onto a normalized output
// //
// This package lets a source emit a "raw" event (e.g., Schema="raw.openweather.current.v1", // Typical usage:
// Payload=json.RawMessage), and then a normalize.Processor can convert it into a // sources emit raw events (often with json.RawMessage payloads), then a
// normalized event (e.g., Schema="weather.observation.v1", Payload=WeatherObservation{}). // normalize.Processor converts matching raw schemas into canonical payloads.
// //
// Key property: normalization is optional. // Key property: normalization is optional.
// If no Normalizer matches an event, Processor passes it through unchanged by default. // If no Normalizer matches an event, Processor passes it through unchanged by default.

7
scheduler/worker.go
View File

@@ -1,7 +0,0 @@
package scheduler
// Placeholder for per-source worker logic:
// - ticker loop
// - jitter
// - backoff on errors
// - emits events into scheduler.Out

33
sinks/builtins.go
View File

@@ -1,11 +1,6 @@
package sinks package sinks
import ( import "gitea.maximumdirect.net/ejr/feedkit/config"
"fmt"
"strings"
"gitea.maximumdirect.net/ejr/feedkit/config"
)
// RegisterBuiltins registers sink drivers included in this binary. // RegisterBuiltins registers sink drivers included in this binary.
// //
@@ -17,11 +12,6 @@ func RegisterBuiltins(r *Registry) {
return NewStdoutSink(cfg.Name), nil return NewStdoutSink(cfg.Name), nil
}) })
// File sink: writes/archives events somewhere on disk.
r.Register("file", func(cfg config.SinkConfig) (Sink, error) {
return NewFileSinkFromConfig(cfg)
})
// Postgres sink: persists events durably. // Postgres sink: persists events durably.
r.Register("postgres", func(cfg config.SinkConfig) (Sink, error) { r.Register("postgres", func(cfg config.SinkConfig) (Sink, error) {
return NewPostgresSinkFromConfig(cfg) return NewPostgresSinkFromConfig(cfg)
@@ -32,24 +22,3 @@ func RegisterBuiltins(r *Registry) {
return NewNATSSinkFromConfig(cfg) return NewNATSSinkFromConfig(cfg)
}) })
} }
// ---- helpers for validating sink params ----
//
// These helpers live in sinks (not config) on purpose:
// - config is domain-agnostic and should not embed driver-specific validation helpers.
// - sinks are adapters; validating their own params here keeps the logic near the driver.
func requireStringParam(cfg config.SinkConfig, key string) (string, error) {
v, ok := cfg.Params[key]
if !ok {
return "", fmt.Errorf("sink %q: params.%s is required", cfg.Name, key)
}
s, ok := v.(string)
if !ok {
return "", fmt.Errorf("sink %q: params.%s must be a string", cfg.Name, key)
}
if strings.TrimSpace(s) == "" {
return "", fmt.Errorf("sink %q: params.%s cannot be empty", cfg.Name, key)
}
return s, nil
}

17
sinks/doc.go
View File

@@ -1,18 +1,27 @@
// Package sinks provides sink abstractions, a sink driver registry, and several // Package sinks defines the feedkit sink interface, sink driver registry, and
// built-in sink drivers. // built-in infrastructure sinks.
//
// External API surface:
// - Sink: adapter interface that consumes event.Event values
// - Registry / NewRegistry: named sink factory registry
// - RegisterBuiltins: registers the built-in sink drivers in this binary
// //
// Built-in drivers: // Built-in drivers:
// - stdout // - stdout
// - nats // - nats
// - postgres // - postgres
// //
// Optional helpers from helpers.go:
// - RegisterPostgresSchemaForConfiguredSinks: registers one Postgres schema
// for each configured sink using driver=postgres
//
// # NATS built-in overview // # NATS built-in overview
// //
// The NATS sink publishes each event as JSON to a configured subject. // The NATS sink publishes each event as JSON to a configured subject.
// //
// Required params: // Required params:
// - url: NATS server URL (for example, nats://localhost:4222) // - url: NATS server URL (for example, nats://localhost:4222)
// - exchange: NATS subject to publish to // - subject: NATS subject to publish to
// //
// Example config: // Example config:
// //
@@ -21,7 +30,7 @@
// driver: nats // driver: nats
// params: // params:
// url: nats://localhost:4222 // url: nats://localhost:4222
// exchange: feedkit.events // subject: feedkit.events
// //
// # Postgres built-in overview // # Postgres built-in overview
// //

30
sinks/file.go
View File

@@ -1,30 +0,0 @@
package sinks
import (
"context"
"fmt"
"gitea.maximumdirect.net/ejr/feedkit/config"
"gitea.maximumdirect.net/ejr/feedkit/event"
)
type FileSink struct {
name string
path string
}
func NewFileSinkFromConfig(cfg config.SinkConfig) (Sink, error) {
path, err := requireStringParam(cfg, "path")
if err != nil {
return nil, err
}
return &FileSink{name: cfg.Name, path: path}, nil
}
func (s *FileSink) Name() string { return s.name }
func (s *FileSink) Consume(ctx context.Context, e event.Event) error {
_ = ctx
_ = e
return fmt.Errorf("file sink: TODO implement (path=%s)", s.path)
}

19
sinks/helpers.go
View File

@@ -7,6 +7,25 @@ import (
"gitea.maximumdirect.net/ejr/feedkit/config" "gitea.maximumdirect.net/ejr/feedkit/config"
) )
// requireStringParam returns a non-empty string sink param.
//
// This helper is intentionally local to sinks rather than config so
// driver-specific validation stays close to the adapters that use it.
func requireStringParam(cfg config.SinkConfig, key string) (string, error) {
v, ok := cfg.Params[key]
if !ok {
return "", fmt.Errorf("sink %q: params.%s is required", cfg.Name, key)
}
s, ok := v.(string)
if !ok {
return "", fmt.Errorf("sink %q: params.%s must be a string", cfg.Name, key)
}
if strings.TrimSpace(s) == "" {
return "", fmt.Errorf("sink %q: params.%s cannot be empty", cfg.Name, key)
}
return s, nil
}
// RegisterPostgresSchemaForConfiguredSinks registers one Postgres schema for each // RegisterPostgresSchemaForConfiguredSinks registers one Postgres schema for each
// configured sink using driver=postgres. // configured sink using driver=postgres.
func RegisterPostgresSchemaForConfiguredSinks(cfg *config.Config, schema PostgresSchema) error { func RegisterPostgresSchemaForConfiguredSinks(cfg *config.Config, schema PostgresSchema) error {

12
sinks/nats.go
View File

@@ -13,9 +13,9 @@ import (
) )
type NATSSink struct { type NATSSink struct {
name string name string
url string url string
exchange string subject string
mu sync.Mutex mu sync.Mutex
conn *nats.Conn conn *nats.Conn
@@ -26,11 +26,11 @@ func NewNATSSinkFromConfig(cfg config.SinkConfig) (Sink, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
ex, err := requireStringParam(cfg, "exchange") subject, err := requireStringParam(cfg, "subject")
if err != nil { if err != nil {
return nil, err return nil, err
} }
return &NATSSink{name: cfg.Name, url: url, exchange: ex}, nil return &NATSSink{name: cfg.Name, url: url, subject: subject}, nil
} }
func (r *NATSSink) Name() string { return r.name } func (r *NATSSink) Name() string { return r.name }
@@ -59,7 +59,7 @@ func (r *NATSSink) Consume(ctx context.Context, e event.Event) error {
if err := ctx.Err(); err != nil { if err := ctx.Err(); err != nil {
return err return err
} }
if err := conn.Publish(r.exchange, b); err != nil { if err := conn.Publish(r.subject, b); err != nil {
return fmt.Errorf("NATS sink: publish: %w", err) return fmt.Errorf("NATS sink: publish: %w", err)
} }
return nil return nil

47
sinks/nats_test.go Normal file
View File

@@ -0,0 +1,47 @@
package sinks
import (
"strings"
"testing"
"gitea.maximumdirect.net/ejr/feedkit/config"
)
func TestNewNATSSinkFromConfigRequiresSubject(t *testing.T) {
sink, err := NewNATSSinkFromConfig(config.SinkConfig{
Name: "nats-main",
Driver: "nats",
Params: map[string]any{
"url": "nats://localhost:4222",
"subject": "feedkit.events",
},
})
if err != nil {
t.Fatalf("NewNATSSinkFromConfig() error = %v", err)
}
natsSink, ok := sink.(*NATSSink)
if !ok {
t.Fatalf("sink type = %T, want *NATSSink", sink)
}
if natsSink.subject != "feedkit.events" {
t.Fatalf("subject = %q, want feedkit.events", natsSink.subject)
}
}
func TestNewNATSSinkFromConfigRejectsLegacyExchange(t *testing.T) {
_, err := NewNATSSinkFromConfig(config.SinkConfig{
Name: "nats-main",
Driver: "nats",
Params: map[string]any{
"url": "nats://localhost:4222",
"exchange": "feedkit.events",
},
})
if err == nil {
t.Fatalf("NewNATSSinkFromConfig() expected error")
}
if !strings.Contains(err.Error(), "params.subject is required") {
t.Fatalf("error = %q, want params.subject is required", err)
}
}

36
sinks/registry.go
View File

@@ -2,6 +2,7 @@ package sinks
import ( import (
"fmt" "fmt"
"strings"
"gitea.maximumdirect.net/ejr/feedkit/config" "gitea.maximumdirect.net/ejr/feedkit/config"
) )
@@ -21,13 +22,40 @@ func NewRegistry() *Registry {
} }
func (r *Registry) Register(driver string, f Factory) { func (r *Registry) Register(driver string, f Factory) {
if r == nil {
panic("sinks.Registry.Register: registry cannot be nil")
}
driver = strings.TrimSpace(driver)
if driver == "" {
panic("sinks.Registry.Register: driver cannot be empty")
}
if f == nil {
panic(fmt.Sprintf("sinks.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("sinks.Registry.Register: driver %q already registered", driver))
}
r.byDriver[driver] = f r.byDriver[driver] = f
} }
func (r *Registry) Build(cfg config.SinkConfig) (Sink, error) { func (r *Registry) Build(cfg config.SinkConfig) (Sink, error) {
f, ok := r.byDriver[cfg.Driver] if r == nil {
if !ok { return nil, fmt.Errorf("sinks registry is nil")
return nil, fmt.Errorf("unknown sink driver: %q", cfg.Driver)
} }
return f(cfg) driver := strings.TrimSpace(cfg.Driver)
f, ok := r.byDriver[driver]
if !ok {
return nil, fmt.Errorf("unknown sink driver: %q", driver)
}
sink, err := f(cfg)
if err != nil {
return nil, fmt.Errorf("build sink %q: %w", driver, err)
}
if sink == nil {
return nil, fmt.Errorf("build sink %q: factory returned nil sink", driver)
}
return sink, nil
} }

126
sinks/registry_test.go Normal file
View File

@@ -0,0 +1,126 @@
package sinks
import (
"context"
"errors"
"strings"
"testing"
"gitea.maximumdirect.net/ejr/feedkit/config"
"gitea.maximumdirect.net/ejr/feedkit/event"
)
type testSink struct{ name string }
func (s testSink) Name() string { return s.name }
func (s testSink) Consume(context.Context, event.Event) error { return nil }
func TestRegistryRegisterPanicsOnNilRegistry(t *testing.T) {
var r *Registry
defer func() {
if recover() == nil {
t.Fatalf("Register() expected panic on nil registry")
}
}()
r.Register("stdout", func(config.SinkConfig) (Sink, error) { return testSink{name: "stdout"}, nil })
}
func TestRegistryRegisterPanicsOnEmptyDriver(t *testing.T) {
r := NewRegistry()
defer func() {
if recover() == nil {
t.Fatalf("Register() expected panic on empty driver")
}
}()
r.Register(" ", func(config.SinkConfig) (Sink, error) { return testSink{name: "x"}, nil })
}
func TestRegistryRegisterPanicsOnNilFactory(t *testing.T) {
r := NewRegistry()
defer func() {
if recover() == nil {
t.Fatalf("Register() expected panic on nil factory")
}
}()
r.Register("stdout", nil)
}
func TestRegistryRegisterPanicsOnDuplicateDriver(t *testing.T) {
r := NewRegistry()
r.Register("stdout", func(config.SinkConfig) (Sink, error) { return testSink{name: "a"}, nil })
defer func() {
if recover() == nil {
t.Fatalf("Register() expected panic on duplicate driver")
}
}()
r.Register("stdout", func(config.SinkConfig) (Sink, error) { return testSink{name: "b"}, nil })
}
func TestRegistryBuildNilRegistryFails(t *testing.T) {
var r *Registry
_, err := r.Build(config.SinkConfig{Driver: "stdout"})
if err == nil {
t.Fatalf("Build() expected error for nil registry")
}
if !strings.Contains(err.Error(), "registry is nil") {
t.Fatalf("Build() error = %q, want registry is nil", err)
}
}
func TestRegistryBuildTrimsDriver(t *testing.T) {
r := NewRegistry()
r.Register("stdout", func(config.SinkConfig) (Sink, error) { return testSink{name: "stdout"}, nil })
sink, err := r.Build(config.SinkConfig{Name: "sink1", Driver: " stdout "})
if err != nil {
t.Fatalf("Build() error = %v", err)
}
if sink.Name() != "stdout" {
t.Fatalf("Build() sink name = %q, want stdout", sink.Name())
}
}
func TestRegistryBuildWrapsFactoryError(t *testing.T) {
r := NewRegistry()
r.Register("broken", func(config.SinkConfig) (Sink, error) { return nil, errors.New("boom") })
_, err := r.Build(config.SinkConfig{Driver: "broken"})
if err == nil {
t.Fatalf("Build() expected error")
}
if !strings.Contains(err.Error(), `build sink "broken": boom`) {
t.Fatalf("Build() error = %q", err)
}
}
func TestRegistryBuildRejectsNilSink(t *testing.T) {
r := NewRegistry()
r.Register("nil_sink", func(config.SinkConfig) (Sink, error) { return nil, nil })
_, err := r.Build(config.SinkConfig{Driver: "nil_sink"})
if err == nil {
t.Fatalf("Build() expected nil sink error")
}
if !strings.Contains(err.Error(), "factory returned nil sink") {
t.Fatalf("Build() error = %q", err)
}
}
func TestRegisterBuiltinsExposesExpectedDrivers(t *testing.T) {
r := NewRegistry()
RegisterBuiltins(r)
if len(r.byDriver) != 3 {
t.Fatalf("len(byDriver) = %d, want 3", len(r.byDriver))
}
for _, driver := range []string{"stdout", "nats", "postgres"} {
if _, ok := r.byDriver[driver]; !ok {
t.Fatalf("builtins missing driver %q", driver)
}
}
if _, ok := r.byDriver["file"]; ok {
t.Fatalf("builtins unexpectedly registered file driver")
}
}

20
sources/doc.go
View File

@@ -1,10 +1,12 @@
// Package sources defines feedkit's input-source abstraction. // Package sources defines feedkit's input-source abstractions and source
// registry.
// //
// A source ingests upstream input and emits one or more event.Event values. // External API surface:
// // - Input: common source identity surface
// feedkit supports two source modes: // - PollSource: polling source interface
// - PollSource: scheduler invokes Poll on a cadence. // - StreamSource: streaming source interface
// - StreamSource: source runs continuously and pushes events as input arrives. // - Registry / NewRegistry: source driver registry and builders
// - HTTPSource / NewHTTPSource: reusable HTTP polling helper
// //
// Source drivers are domain-specific and registered into Registry by driver name. // Source drivers are domain-specific and registered into Registry by driver name.
// Registry can then build configured sources from config.SourceConfig. // Registry can then build configured sources from config.SourceConfig.
@@ -12,6 +14,12 @@
// A single source may emit 0..N events per poll or stream iteration, and those // A single source may emit 0..N events per poll or stream iteration, and those
// events may span multiple event kinds. // events may span multiple event kinds.
// //
// Optional helpers from helpers.go:
// - DefaultEventID: default event ID policy for source implementations
// - SingleEvent: construct and validate a one-element event slice
// - ValidateExpectedKinds: compare configured expected kinds against source
// advertised kinds when metadata is available
//
// HTTP-backed polling sources can share NewHTTPSource for generic HTTP config // HTTP-backed polling sources can share NewHTTPSource for generic HTTP config
// parsing and conditional GET behavior. The helper understands: // parsing and conditional GET behavior. The helper understands:
// - params.url // - params.url

35
sources/http.go
View File

@@ -5,7 +5,6 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http" "net/http"
"strconv"
"strings" "strings"
"sync" "sync"
@@ -60,9 +59,13 @@ func NewHTTPSource(driver string, cfg config.SourceConfig, accept string) (*HTTP
return nil, fmt.Errorf("%s %q: params.user_agent is required", driver, cfg.Name) return nil, fmt.Errorf("%s %q: params.user_agent is required", driver, cfg.Name)
} }
conditional, err := parseConditionalParam(cfg) conditional := true
if err != nil { if _, exists := cfg.Params["conditional"]; exists {
return nil, err var ok bool
conditional, ok = cfg.ParamBool("conditional")
if !ok {
return nil, fmt.Errorf("source %q: params.conditional must be a boolean", cfg.Name)
}
} }
return &HTTPSource{ return &HTTPSource{
@@ -121,27 +124,3 @@ func (s *HTTPSource) FetchJSONIfChanged(ctx context.Context) (json.RawMessage, b
} }
return json.RawMessage(body), true, nil 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)
}
}

22
sources/http_test.go
View File

@@ -4,6 +4,7 @@ import (
"context" "context"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"strings"
"testing" "testing"
"gitea.maximumdirect.net/ejr/feedkit/config" "gitea.maximumdirect.net/ejr/feedkit/config"
@@ -39,6 +40,27 @@ func TestNewHTTPSourceRejectsInvalidConditional(t *testing.T) {
if err == nil { if err == nil {
t.Fatalf("NewHTTPSource() error = nil, want error") t.Fatalf("NewHTTPSource() error = nil, want error")
} }
if !strings.Contains(err.Error(), "params.conditional must be a boolean") {
t.Fatalf("NewHTTPSource() error = %q, want params.conditional must be a boolean", err)
}
}
func TestNewHTTPSourceConditionalCanBeExplicitlyFalse(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",
"conditional": false,
},
}, "application/json")
if err != nil {
t.Fatalf("NewHTTPSource() error = %v", err)
}
if src.Conditional {
t.Fatalf("Conditional = true, want false")
}
} }
func TestHTTPSourceFetchJSONIfChanged(t *testing.T) { func TestHTTPSourceFetchJSONIfChanged(t *testing.T) {