insights

Documentation

Overview

Package insights projects opt-in domain entities into each tenant's own fabriq_insights_facts table (per-tenant customer-facing analytics), version- gated and RLS-scoped. It mirrors core/analytics (the cross-tenant operator sink) but does NOT redact — the data never leaves the tenant's own database — and it does not write Events/Watermark rows (phase 1 is facts only).

Dependency fence: this package imports only core/event, core/registry, core/projection, core/tenant, and internal/otel. It must NOT import adapters/* (the postgres adapter imports this package to implement FactSink — the reverse direction would be an import cycle).

Package insights holds the behavioral contract for the per-tenant, customer-facing analytics port (query.AnalyticsQuerier). It is deliberately distinct from core/analytics, which is the operator-facing cross-tenant sink conformance suite for a different port (analytics.Sink).

Index

• func EffectiveQuery(q query.AnalyticsQuery, d Descriptor) (measures []query.Measure, dimensions []string, bucket time.Duration, err error)
• func RunConformance(t *testing.T, newQ func(reg *registry.Registry) query.AnalyticsQuerier)
• type Applier
  • func NewApplier(reg *registry.Registry) *Applier
  • func (a *Applier) Apply(env event.Envelope) (Fact, bool, error)
• type Consumer
  • func (c *Consumer) Run(ctx context.Context, name string) error
• type Descriptor
  • func ResolveSource(reg *registry.Registry, source string) (Descriptor, error)
• type Fact
• type FactSink
• type RollupSurface
• type SourceKind

Functions

func EffectiveQuery

func EffectiveQuery(q query.AnalyticsQuery, d Descriptor) (measures []query.Measure, dimensions []string, bucket time.Duration, err error)

EffectiveQuery computes the measures/dimensions/bucket a query actually runs with, given the Descriptor ResolveSource produced for q.Source. Both buildInsightsSQL (adapters/postgres) and the in-memory fake (fabriqtest.FakeAnalytics) call this — never q.Measures/q.Dimensions/ q.TimeBucket directly — so metric expansion can't drift between the two.

When d.FromMetric (q.Source named a declared MetricSpec), the metric's own Measures/Dimensions/DefaultBucket apply and the caller may NOT also pass Measures or Dimensions (that would be ambiguous: which wins?) — doing so is rejected outright. q.TimeBucket, when set, still overrides the metric's DefaultBucket (a caller-supplied window granularity is not ambiguous the same way). Filter/Having/From/To/OrderBy/Limit/Offset are ALWAYS the caller's — EffectiveQuery has no opinion on them; it only resolves the three fields a metric declares.

When d.FromMetric is false, q.Measures/q.Dimensions/q.TimeBucket pass through unchanged.

func RunConformance

func RunConformance(t *testing.T, newQ func(reg *registry.Registry) query.AnalyticsQuerier)

RunConformance is the single behavioral contract every query.AnalyticsQuerier must satisfy. fabriqtest runs it against FakeAnalytics; adapters/postgres runs it against real Postgres. Drift is a failing test.

It exercises Track + Query only. QueryRaw has no portable in-memory contract (raw SQL is dialect-specific): the adapter's own test suite exercises QueryRaw separately.

newQ is invoked once per sub-test (some fakes truncate state per-call) and is handed the suite's registry so it can wire the querier under test (fake or real adapter) with the same reg every time — the registry the insights.ResolveSource routing decisions are made against. The suite's own existing event-only subtests need nothing registered in reg; Tasks 6/7 add entities/metrics to it for the facts/metric subtests they introduce.

Types

type Applier

type Applier struct {
	// contains filtered or unexported fields
}

Applier turns one event.Envelope into a Fact using the entity's registry InsightsSpec. It is PURE and deterministic: the same envelope always yields the same bytes (so live apply and backfill agree). Deny-by-default: entities without an InsightsSpec produce no records.

func NewApplier

func NewApplier(reg *registry.Registry) *Applier

func (*Applier) Apply

func (a *Applier) Apply(env event.Envelope) (Fact, bool, error)

Apply returns (fact, ok, err). ok=false means the entity is not Insights- projected — the caller skips it. No redaction (the tenant owns its own data); the payload is projected to the declared Measures+Dimensions columns only, keeping facts narrow. Canonical (sorted-key) JSON so live apply and backfill agree byte-for-byte.

type Consumer

type Consumer struct {
	Group     string // "proj:insights"
	Source    projection.Source
	Applier   *Applier
	Sink      FactSink
	Upcasters *event.UpcasterChain // optional

	// OnApplied and OnFailure are optional observability hooks, nil-safe.
	// Neither fires on the skip path (un-migratable payload, malformed
	// payload, un-derivable tenant, or an unmarked entity).
	OnApplied func()
	OnFailure func()
}

Consumer drives the proj:insights group: it reads the shared event stream through the exported projection.Source seam and applies each envelope to the FactSink. Phase 1 projects FACTS ONLY — no Events/Watermark rows (those are the operator-sink's concern, not the tenant's own store). Idempotency lives in the Sink (version gate), so redelivery is always safe.

func (*Consumer) Run

func (c *Consumer) Run(ctx context.Context, name string) error

Run consumes until ctx ends. Scale by running replicas with distinct names.

type Descriptor

type Descriptor struct {
	// Kind is SourceEvent or SourceFacts.
	Kind SourceKind
	// Table is the physical table to read: fabriq_insights_events or
	// fabriq_insights_facts.
	Table string
	// JSONColumn is the JSONB column holding the schemaless payload: "props"
	// for events, "payload" for facts.
	JSONColumn string
	// KeyColumn is the column that scopes rows to this source: "name" for
	// events, "entity" for facts.
	KeyColumn string
	// KeyValue is the value KeyColumn is filtered to (the event name or the
	// projected entity name).
	KeyValue string
	// ExtraWhere is an additional literal SQL predicate ANDed into every query
	// against this source (e.g. "deleted = false" for facts, to hide
	// tombstoned rows). Empty for events.
	ExtraWhere string

	// FromMetric is true when the original Query.Source named a declared
	// MetricSpec rather than an event or entity directly.
	FromMetric bool
	// MetricName is the declared metric's name (set only when FromMetric).
	// Present for error messages that need to name the metric distinctly from
	// its underlying Source.
	MetricName string
	// MetricMeasures are the metric's declared measures, translated to
	// query.Measure (set only when FromMetric).
	MetricMeasures []query.Measure
	// MetricDimensions are the metric's declared dimensions (set only when
	// FromMetric).
	MetricDimensions []string
	// MetricBucket is the metric's declared default time bucket, zero if
	// none (set only when FromMetric).
	MetricBucket time.Duration

	// AllowedColumns, for SourceFacts, is the allow-list of columns (measures
	// ∪ dimensions) a query against this entity may reference — the
	// InsightsSpec column allow-list. Nil for SourceEvent (event props are
	// schemaless; any top-level key is allowed).
	AllowedColumns map[string]bool
}

Descriptor is the resolved shape of an AnalyticsQuery.Source: which table to read, how to key into it, and (for metric sources) the measures/ dimensions/bucket the metric declares. Pure data — no I/O, no SQL. Both the postgres cube builder and the in-memory fake consume the same Descriptor so routing can't drift between them.

func ResolveSource

func ResolveSource(reg *registry.Registry, source string) (Descriptor, error)

ResolveSource decides whether source is a declared metric (expand it to its underlying event/facts source plus its measures/dimensions), a projected entity (read facts), or a customer event (read events). Precedence is metric > entity > event. Pure; no I/O.

A nil registry resolves everything to an event descriptor — the back-compat mode the in-memory fake supports when no registry is wired.

type Fact

type Fact struct {
	TenantID string
	Entity   string
	AggID    string
	Version  int64
	Payload  json.RawMessage // the entity payload projected to Insights columns
	At       time.Time
	Deleted  bool
}

Fact is one projected latest-state row for an opt-in domain entity.

type FactSink

type FactSink interface {
	UpsertInsightFacts(ctx context.Context, facts []Fact) error
}

FactSink upserts projected facts into the tenant's own store, version- gated. Implemented by the postgres adapter (writes fabriq_insights_facts through inTenantTx, so RLS contains it). ctx carries the tenant.

type RollupSurface

type RollupSurface interface {
	EnsureRollupTable(ctx context.Context, m *registry.MetricSpec) error
	MaintainRollup(ctx context.Context, m *registry.MetricSpec, now time.Time) error
}

RollupSurface is the per-tenant Postgres rollup-maintenance capability the leader-elected rollup:insights maintainer job (forgeext) needs:

• EnsureRollupTable creates (idempotently) a materialized metric's rollup table via the schema-owner DDL path.
• MaintainRollup runs one seal-aggregate-advance-watermark pass for a materialized metric under the caller's (unscoped) tenant ctx.

Implemented by *postgres.InsightsAdapter (a thin pass-through to the underlying *postgres.Adapter), resolved per tenant through the same shard/router seam FactSink uses (Shard.Analytics) — so a maintainer pass always lands on the tenant's own shard (static sharding) or tenant database (catalog / db-per-tenant mode), exactly like proj:insights' per-event writes.

type SourceKind

type SourceKind int

SourceKind distinguishes the two physical stores an AnalyticsQuery.Source can resolve to: raw customer events (fabriq_insights_events) or projected entity facts (fabriq_insights_facts).

const (
	// SourceEvent means the query reads the tenant's fabriq_insights_events
	// table, keyed by event name.
	SourceEvent SourceKind = iota
	// SourceFacts means the query reads the tenant's fabriq_insights_facts
	// table, keyed by projected entity name.
	SourceFacts
)