Analytics sink

Every fabriq tenancy mode is built around one invariant: no cross-tenant queries. That is exactly why fleet-wide reporting ("orders fleet-wide, by status") has nowhere to run — not in a shared database, not across shards, not across thousands of per-tenant databases — without punching a hole in the isolation the rest of the system guarantees. The analytics sink is the one deliberate, narrow, auditable exception: an opt-in projection that co-locates a redacted, allow-listed subset of tenant data for operator-only reporting.

Topology01

The analytics sink is a third projection peer to graph and search — it reads the same shared event stream, not a new transport:

proj:analytics is an ordinary Redis consumer group on registry.StreamKey(), consumed through the exported projection.Source seam (core/analytics.Consumer deliberately does not wrap projection.Engine, which is hardwired to graph/search's closed Mutation set). Each envelope is turned into a redacted Fact + Event by a pure core/analytics.Applier, then written through the core/analytics.Sink port. The reference adapter, adapters/pganalytics, is a shared Postgres database.

Because it rides the existing stream, the analytics sink starts in every tenancy mode — static and catalog alike — with no per-shard or per-tenant-database wiring.

Enabling it02

The sink is off unless Config.Analytics.DSN is set, and the proj:analytics consumer only starts when a DSN is configured AND at least one entity opts in:

analytics:
  dsn: postgres://fabriq_analytics@analytics-db:5432/fabriq_analytics
  batch: 128   # backfill write batch size (default 128)

The analytics DSN must differ from every tenant, shard, and catalog control DSN — ValidateAnalyticsConfig checks this at fabriq.Open and refuses to start on a collision. Separate database, separate credential, enforced at boot, not left as an operational convention.

Mark an aggregate for analytics with an allow-list of payload fields on its registry.EntitySpec:

r.Register(registry.EntitySpec{
    Name: "order",
    // ...
    Analytics: &registry.AnalyticsSpec{
        Include: []string{"status", "category", "total"},
    },
})

Everything not named in Include is stripped before the record leaves the projection — deny-by-default at the aggregate level, then field minimization within a marked aggregate. Field names may be dot-paths ("meta.region") that descend through nested JSON objects, so you can keep or hash a single nested leaf and drop its PII siblings; a path whose intermediate segment is missing or isn't an object simply contributes nothing. Leaf values keep their exact bytes (numbers and ids never lose precision). IncludeAll: true ships the whole payload unredacted; use it only when you have already reviewed the payload for PII. A spec with no Include, Hash, or IncludeAll is rejected by the registry (a marked-but-empty spec would silently analyticize nothing).

Pseudonymizing a field

Sometimes you need to analyze by a sensitive field — count distinct users across the fleet, group orders by customer — without co-locating the raw value. List those fields under Hash: their value is replaced with a stable, salted hash before it crosses the boundary. Equal values hash equally (so group-by and count-distinct still work), but the raw value never lands and is not recoverable.

Analytics: &registry.AnalyticsSpec{
    Include: []string{"status", "total"},
    Hash:    []string{"customerId"},   // pseudonymized, not raw
},

The hash uses Config.Analytics.HashSalt — a stable, secret, deployment-wide salt (a leaked salt makes the hashes brute-forceable). It is required when any entity uses Hash; Open fails fast otherwise. Because the salt is stable, live apply, backfill, and reproject all produce the same hash for the same value. A hashed field is implicitly included — it need not also appear in Include.

Choosing a backend03

Config.Analytics.DSN's URL scheme selects the sink adapter at Open — there is no separate driver setting:

An unknown scheme fails fast at boot: fabriq: unknown analytics DSN scheme "…". All three adapters satisfy the same core/analytics.Sink port and run through the identical conformance suite, so they are drop-in interchangeable at that level — the differences below are operational, not behavioral.

Postgres (default)

The reference adapter described throughout this page. It is the only backend that supports partitionEvents (see Partitioning at scale) — DuckDB and ClickHouse don't have a native partition-drop primitive wired into the sink, so partitionEvents: true with a non-postgres DSN scheme is also a boot error: fabriq: analytics PartitionEvents is only supported on the postgres backend.

ClickHouse

analytics:
  dsn: clickhouse://user:pass@analytics-db:9000/fabriq_analytics

The three tables are ReplacingMergeTree, ClickHouse's columnar merge-on-read engine — there is no in-place UPDATE, so the version-gate described above is enforced by keeping the highest-versioned row per key and letting background merges (or a query-time FINAL/argMax) resolve to the winner. This changes how you read the store directly:

• Point/aggregate reads issued by the sink itself never need FINAL — they use argMax(col, _dedup) (facts/events) or plain max(version) (watermarks), which is correct without waiting for a background merge.

• Ad hoc operator queries against fabriq_analytics_facts or fabriq_analytics_events should add FINAL to the FROM clause (or aggregate with argMax the same way) — without it, a query can see both a stale and a current row for the same key until the next merge.

• payload is stored as String (ClickHouse's native JSON type is still experimental), so query into it with JSONExtract* functions, e.g. JSONExtractString(payload, 'status'), rather than a JSONB-style ->>.

• Prune and purge (fabriq analytics prune-events, PurgeTenant) issue a lightweight DELETE FROM … WHERE …, generally available since ClickHouse 23.3. On an older server, enable allow_experimental_lightweight_delete or upgrade before relying on those operations.

The ClickHouse driver is pure Go — it ships in the default CGO_ENABLED=0 release binary with no extra build tag.

DuckDB

analytics:
  dsn: duckdb:///var/lib/fabriq/analytics.db   # file-backed
  # dsn: duckdb://:memory:                     # in-process, non-durable

DuckDB is embedded — no server to run, no network hop — and columnar like ClickHouse, but with full mutable SQL (UPDATE/DELETE in place), so its adapter reads closer to the Postgres one than the ClickHouse one.

Because it's in-process, a DuckDB sink is tied to the fabriq instance's own disk and lifetime — there's no separate database to point a second reader at. It suits a single-node deployment or local/edge reporting more than a shared fleet-wide store queried by other tools; for that, ClickHouse or Postgres remain the better fit.

Schema04

adapters/pganalytics ensures its schema idempotently at Open (three CREATE TABLE IF NOT EXISTS statements) rather than joining fabriq's tenant migration chain — this is a self-contained operator store:

Every table has tenant_id NOT NULL. There is no RLS on this database — see the trust boundary callout above.

The version gate on fabriq_analytics_facts is what turns Redis's at-least-once delivery into an idempotent effect: redelivery of an already-applied envelope is a no-op, with no separate dedupe table or lock required. The steady-state apply path only ever reads and writes the analytics database — it never touches a tenant database.

Query patterns05

Every query against the analytics database filters or groups by tenant_id — that is the whole point of retaining the column even though there is no RLS to enforce it. Fleet-wide aggregation groups across tenants deliberately; per-tenant reporting still filters by it.

Fleet-wide count of live orders by status:

SELECT payload->>'status' AS status, count(*)
FROM fabriq_analytics_facts
WHERE aggregate = 'order' AND NOT deleted
GROUP BY payload->>'status';

Per-tenant history for one order:

SELECT version, type, payload, at
FROM fabriq_analytics_events
WHERE tenant_id = $1 AND aggregate = 'order' AND agg_id = $2
ORDER BY version;

Backfill runbook06

Backfill replays a tenant's current-state snapshot — not the event log — through the same Applier the live consumer uses, so a re-run is a no-op (version-gated) and safe to run alongside live traffic. Use it when you mark a new entity for analytics (its history predates the marking) or to recover a dropped/rebuilt analytics database.

CLI, one tenant or the whole fleet:

export FABRIQ_ANALYTICS_DSN=postgres://fabriq_analytics@analytics-db:5432/fabriq_analytics

fabriq analytics backfill --tenant acme
fabriq analytics backfill --all-tenants --concurrency 8

With the admin extension mounted and WithAnalyticsAdmin() enabled, the analytics.admin/analytics.read capabilities expose the same operation over HTTP:

Async for large fleets. A fleet-wide op ("all": true) on a big fleet can run longer than an HTTP timeout. Add "async": true to backfill, reproject, or reconcile and the endpoint returns 202 with a {"jobId"} immediately; poll GET {base}/analytics/jobs/:id for {state, result} (running → done/failed), or stream state changes over SSE at GET {base}/analytics/jobs/:id/stream. Single-tenant ops stay synchronous.

A synchronous fleet backfill ("all": true) that partially fails returns HTTP 207 with the partial per-tenant counts and an error string, rather than discarding successful tenants' results. Backfill is deliberately synchronous today — an async job/SSE variant (matching the tenant provisioning and migration job APIs) is a documented future enhancement, not built here.

Erasure and offboarding07

Because the sink co-locates many tenants' data, it needs an explicit way to remove a tenant — for offboarding and for right-to-be-forgotten requests. PurgeTenant hard-deletes all of one tenant's rows across the three tables (facts, events, and watermarks) in a single transaction and reports the count removed. It is idempotent (purging an absent tenant deletes nothing) and is the only supported way to remove a tenant from the store.

# CLI — the destructive erase requires an explicit --yes.
fabriq analytics purge --tenant acme --yes

Changing redaction after the fact08

Tightening an entity's Include allow-list (or dropping a field) does not retroactively strip that field from rows already in the sink: backfill replays at each row's current version, and the fact upsert is version-gated, so a same-version replay is a no-op — and the event log is immutable. Reprojection closes that gap. It re-projects every stored fact and event payload through the entity's current spec, in place, rewriting only rows that actually change:

fabriq analytics reproject --tenant acme
fabriq analytics reproject --all-tenants --concurrency 8

Event-log retention09

fabriq_analytics_facts holds one latest-state row per aggregate and stays bounded, but fabriq_analytics_events is an append-only history that grows without limit. Set a retention window to bound it — a worker pruner then deletes history events older than the window on an hourly cadence (facts are never pruned):

analytics:
  dsn: postgres://…/analytics
  eventRetention: 2160h   # keep 90 days of history; 0 (default) keeps forever

The pruner runs in both the static worker and the catalog sweeper planes and emits fabriq_analytics_events_pruned_total. For a one-off trim (or when no worker runs), use the CLI:

fabriq analytics prune-events --older-than 2160h

Partitioning at scale

At a very large fleet, deleting expired rows one-by-one is expensive. Set partitionEvents to create the event log as a monthly range-partitioned table: retention then reclaims space by dropping whole partitions (instant) instead of a delete-scan, and queries prune irrelevant months automatically.

analytics:
  dsn: postgres://…/analytics
  eventRetention: 2160h
  partitionEvents: true   # monthly partitions; drop-partition retention

A worker maintainer creates upcoming month partitions ahead of the write clock and drops any partition entirely older than the retention window. A default partition catches any write outside the maintained window (e.g. a backfill appending an event with a historical timestamp), so writes never fail; the row-level pruner sweeps that default partition for the boundary.

Reconciling drift10

The consumer skips any event it cannot upcast or apply (poison-avoidance), so a deployment bug can leave a fact permanently missing or stale — the same way it would for the graph and search projections, which heal from Postgres via their reconcilers. Reconciliation is the analytics equivalent: it reads each marked aggregate's current source state, compares it against the stored watermark, and re-applies only the aggregates that are missing or behind.

fabriq analytics reconcile --tenant acme
fabriq analytics reconcile --all-tenants --concurrency 8

Each tenant reports checked / drifted (missing, stale) / healed. It is idempotent — a healthy tenant reports zero drift — and safe to run alongside live traffic (the same version gate protects it). Unlike backfill, which re-applies every row blindly, reconcile only touches what actually diverged, so it doubles as a drift audit: a non-zero drifted count is a signal that the consumer skipped events (check fabriq_analytics_failures_total and the logs).

Freshness, lag, and metrics11

The sink is near-real-time: lag is relay latency plus apply time, sub-second in a healthy deployment, because it rides the same stream graph and search already consume from. Two counters, defined in internal/metrics, are emitted by the proj:analytics consumer: