Documentation
¶
Overview ¶
Package postgres implements usage.Sink, usage.Reader, and usage.Closer backed by PostgreSQL. Events are buffered in-memory and flushed in batches via pgx CopyFrom. PostgreSQL is its own durable store (its own WAL); there is no secondary JSONL segment queue here. On flush error the batch is dropped and counted in Dropped() — best-effort, no re-buffer.
Expected kv ops per request: none (all async). One CopyFrom per flush interval across BatchSize events.
Index ¶
- type Config
- type Sink
- func (s *Sink) Close() error
- func (s *Sink) Dropped() uint64
- func (s *Sink) Events(ctx context.Context, q usage.EventQuery) ([]usage.Event, error)
- func (s *Sink) Summary(ctx context.Context, q usage.SummaryQuery) (usage.SummaryResult, error)
- func (s *Sink) TimeSeries(ctx context.Context, q usage.TimeSeriesQuery) (usage.TimeSeriesResult, error)
- func (s *Sink) Write(ev usage.Event) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// DSN is the PostgreSQL connection string (postgres://user:pass@host/db).
DSN string
// RetentionDays controls how long rows are kept by the daily prune job.
// <=0 disables pruning. Default 90.
RetentionDays int
// BatchSize is the number of events to accumulate before flushing.
// Default 500.
BatchSize int
// FlushInterval is how often buffered events are flushed even if
// BatchSize is not reached. Default 2s.
FlushInterval time.Duration
// Table is the table name. Must match ^[a-z_][a-z0-9_]*$.
// Default "usage_events".
Table string
}
Config holds all tunables for the PostgreSQL sink.
type Sink ¶
type Sink struct {
// contains filtered or unexported fields
}
Sink is the PostgreSQL-backed implementation of usage.Sink, usage.Reader, and usage.Closer.
func New ¶
New opens a pgxpool connection, validates the table name, ensures the schema exists, and starts the background flush ticker.
func (*Sink) Summary ¶
func (s *Sink) Summary(ctx context.Context, q usage.SummaryQuery) (usage.SummaryResult, error)
Summary returns aggregated rows grouped by q.GroupBy. Token sums are computed by unnesting the jsonb map in SQL; latency percentiles use percentile_cont. Both queries share the same WHERE and are merged in Go.
func (*Sink) TimeSeries ¶
func (s *Sink) TimeSeries(ctx context.Context, q usage.TimeSeriesQuery) (usage.TimeSeriesResult, error)
TimeSeries buckets matched events by q.Interval (date_bin, epoch-aligned), optionally split into one series per q.GroupBy value. Like Summary, token sums come from a second jsonb_each_text query merged in Go. Empty buckets are not emitted; series are ordered by total requests desc, points oldest-first.
Source Files