Eric Rakestraw
09bc65e947
- Add logging.Logf as the canonical printf-style logger type used across feedkit.
- Update scheduler and dispatch to alias their Logger types to logging.Logf.
- Eliminates type-mismatch friction when wiring one log function through the system.
- Add dispatch.CompileRoutes(*config.Config) ([]dispatch.Route, error)
- Compiles config routes into dispatch routes with event.ParseKind normalization.
- If routes: is omitted, defaults to “all sinks receive all kinds”.
- Expand config param helpers for both SourceConfig and SinkConfig
- Add ParamBool/ParamInt/ParamDuration/ParamStringSlice (+ Default variants).
- Supports common YAML-decoded types (bool/int/float/string, []any, etc.)
- Keeps driver code cleaner and reduces repeated type assertions.
- Fix Postgres sink validation error prefix ("postgres sink", not "rabbitmq sink").
38 lines
828 B
Go
38 lines
828 B
Go
package sinks
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
|
|
"gitea.maximumdirect.net/ejr/feedkit/config"
|
|
"gitea.maximumdirect.net/ejr/feedkit/event"
|
|
)
|
|
|
|
type PostgresSink struct {
|
|
name string
|
|
dsn string
|
|
}
|
|
|
|
func NewPostgresSinkFromConfig(cfg config.SinkConfig) (Sink, error) {
|
|
dsn, err := requireStringParam(cfg, "dsn")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &PostgresSink{name: cfg.Name, dsn: dsn}, 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.
|
|
if err := e.Validate(); err != nil {
|
|
return fmt.Errorf("postgres sink: invalid event: %w", err)
|
|
}
|
|
|
|
// TODO implement Postgres transaction
|
|
return nil
|
|
}
|