Documentation
¶
Overview ¶
Package pganalytics is the reference analytics.Sink: a shared Postgres analytics database with a tenant_id column on every table. It is the ONE place fabriq deliberately co-locates data from many tenants; the database and its credential are separate from tenant DBs and the catalog control DB.
Index ¶
- func TruncateForTest(ctx context.Context, s *Sink) error
- type Option
- type Sink
- func (s *Sink) AllWatermarks(ctx context.Context, tenantID string) ([]analytics.Watermark, error)
- func (s *Sink) AppendEvents(ctx context.Context, events []analytics.Event) error
- func (s *Sink) Close() error
- func (s *Sink) LagByTenant(ctx context.Context) (map[string]float64, error)
- func (s *Sink) MaintainPartitions(ctx context.Context, retention time.Duration) (int, int, error)
- func (s *Sink) PruneEvents(ctx context.Context, olderThan time.Time) (int64, error)
- func (s *Sink) PurgeTenant(ctx context.Context, tenantID string) (int64, error)
- func (s *Sink) QueryReadOnly(ctx context.Context, query string, args ...any) (rows []map[string]any, cols []string, truncated bool, err error)
- func (s *Sink) ReprojectTenant(ctx context.Context, tenantID, aggregate string, ...) (int64, error)
- func (s *Sink) SetWatermark(ctx context.Context, ws []analytics.Watermark) error
- func (s *Sink) UpsertFacts(ctx context.Context, facts []analytics.Fact) error
- func (s *Sink) Watermark(ctx context.Context, tenantID, aggregate, aggID string) (int64, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type Option ¶
type Option func(*Sink)
Option configures the sink at Open.
func WithEventPartitioning ¶
func WithEventPartitioning() Option
WithEventPartitioning creates the event log as a monthly range-partitioned table (PARTITION BY RANGE (at)) so retention becomes an instant DROP PARTITION instead of a delete-scan. Takes effect only on a FRESH database — CREATE TABLE IF NOT EXISTS will not convert an existing non-partitioned events table; migrating an existing deployment is a manual step.
type Sink ¶
type Sink struct {
// contains filtered or unexported fields
}
Sink writes the analytics read model. It ensures its own schema at Open (three CREATE TABLE IF NOT EXISTS, mirroring the catalog control DB) rather than joining fabriq's tenant migration chain — this is a self-contained operator store.
func Open ¶
Open dials the analytics database and ensures the schema. ensureSchema proves reachability (it requires a live connection).
func (*Sink) AllWatermarks ¶
AllWatermarks returns every applied watermark for a tenant in one read.
func (*Sink) AppendEvents ¶
AppendEvents dedupes on the natural key. The conflict target must match the events primary key, which includes `at` in partitioned mode (`at` is fixed per version, so dedup semantics are unchanged).
func (*Sink) LagByTenant ¶
LagByTenant reports now() - (that tenant's newest fact commit time), in seconds, per tenant. An empty map means no facts. Same failed-iteration-is-an-error discipline as Watermark.
func (*Sink) MaintainPartitions ¶
MaintainPartitions (partitioned mode only) ensures the current and next two months have partitions and, when retention > 0, drops every monthly partition whose entire range is older than now-retention — the instant-reclaim path that replaces a delete-scan. Returns (created, dropped). A no-op when the sink is not partitioned.
func (*Sink) PruneEvents ¶
PruneEvents deletes history events with at < olderThan across all tenants (the (at) index carries the scan) and returns the count removed. Facts are untouched. Idempotent.
func (*Sink) PurgeTenant ¶
PurgeTenant hard-deletes all of one tenant's rows across the three analytics tables and returns the total removed. Runs in one transaction so a tenant is never left half-erased. Idempotent.
func (*Sink) QueryReadOnly ¶
func (s *Sink) QueryReadOnly(ctx context.Context, query string, args ...any) (rows []map[string]any, cols []string, truncated bool, err error)
QueryReadOnly runs a query inside a READ ONLY transaction against the Postgres analytics store and returns a dynamic result set. The read-only tx is the REAL enforcement — a data-modifying statement (including a data-modifying CTE) errors at the database, regardless of the caller's precheck. The row cap bounds the result; the caller bounds time via ctx.
func (*Sink) ReprojectTenant ¶
func (s *Sink) ReprojectTenant(ctx context.Context, tenantID, aggregate string, transform func(json.RawMessage) (json.RawMessage, error)) (int64, error)
ReprojectTenant re-writes stored fact and event payloads for a tenant (and optional aggregate) through transform, in place. Change detection is semantic: the UPDATE only fires when the re-projected JSONB actually differs (payload IS DISTINCT FROM), so RowsAffected counts real rewrites and a re-run is a no-op — the fake compares its canonical bytes and reaches the same count.
func (*Sink) SetWatermark ¶
SetWatermark advances monotonically.
func (*Sink) UpsertFacts ¶
UpsertFacts version-gates: a row is updated only when the incoming version is strictly greater than the stored one.
Source Files