Analytics & warehouse sinks

Voltro ships a narrow, cross-provider analytics contract (AnalyticsSink) and five first-party plugins that implement it. Mirror of the auth-strategy pattern: one typed interface, multiple providers, swap them in app.config.ts without touching handler code.

The intent is "go very far before you reach for an external tool" — but make external tools a one-line install when you do. The first-party lite plugin (postgres-analytics) works on day 1 with zero external services and carries you up to ~10M events/day before queries get slow.

#The contract — AnalyticsSink

Four methods. Effect-typed throughout. Every sink implements all four; sinks that genuinely can't support a capability return AnalyticsCapabilityNotSupported on the typed error channel rather than throwing.

import { useAnalytics } from '@voltro/runtime'

export default (input, ctx) => Effect.gen(function* () {
  const analytics = yield* useAnalytics()

  yield* analytics.track({
    name: 'match_completed',
    subjectId: ctx.request.subject.id,
    properties: { mapId: 'dust2', durationSec: 1284, mvp: 'player_abc' },
  })

  // Total over a window
  const total = yield* analytics.aggregate({
    event:  'match_completed',
    metric: 'count',
    range:  { from: hoursAgo(24) },
  })

  // Time-bucketed series
  const daily = yield* analytics.timeseries({
    event:  'match_completed',
    metric: 'count',
    bucket: 'day',
    range:  { from: daysAgo(30) },
  })

  // Top-N by a property
  const topMaps = yield* analytics.topN({
    event:   'match_completed',
    groupBy: 'mapId',
    metric:  'count',
    n:       10,
    range:   { from: daysAgo(30) },
  })
})

The contract is deliberately narrow. No raw SQL, no funnels, no cohorts, no custom dashboards. Anything provider-specific lives outside the contract — call the provider's API/client directly where you need it (the plugins expose no raw-client escape hatch).

#Plugins at a glance

Plugin	Type	Best for	Ceiling
@voltro/plugin-analytics-postgres	Lite	Day-1 zero-setup, dev + early production	~10M events/day
@voltro/plugin-duckdb	Embedded OLAP	Real column-store performance, no external service	Vertical scale: ~hundreds of GB in one process
@voltro/plugin-clickhouse	External OLAP	Production-scale analytics, self-hosted or ClickHouse Cloud	Billions of events comfortably
@voltro/plugin-tinybird	Hosted ClickHouse	Pay-as-you-go without operating ClickHouse	Tinybird's own limits
@voltro/plugin-posthog	Product analytics	Sessions, feature flags, funnels in PostHog's UI	track() only — compose with another sink for reads

#Picking one

Decision rubric in order:

1. First service you don't want to operate? → postgres-analytics. Works against the main DataStore. Cross-dialect.
2. First-party experience but real OLAP performance? → duckdb. Embedded sidecar, no external service to run.
3. Past 10M events/day OR want billions-of-rows queries to stay under a second? → clickhouse (self-hosted or Cloud) or tinybird (hosted).
4. Already invested in PostHog for product analytics? → posthog for ingest (alongside a primary sink for aggregates).

Switching providers later is a one-line config change. Handlers stay identical because they consume the same AnalyticsSink contract.

#@voltro/plugin-analytics-postgres

First-party lite. Stores events in _voltro_events on the main DataStore. Cross-dialect: works on postgres / mysql / mariadb / mssql / sqlite / turso via sql.onDialectOrElse.

// app.config.ts
import { postgresAnalytics } from '@voltro/plugin-analytics-postgres'

export default defineApi({
  name:      'myApi',
  store:     'postgres',
  analytics: postgresAnalytics(),
})

What the plugin owns:

• The _voltro_events table — auto-created on first boot (numeric id, name, subject_id, properties JSONB, occurred_at). Two composite indexes (name+occurred_at, subject_id+occurred_at) for the hot read paths.
• All four methods: track writes a row; aggregate / timeseries / topN compile to date_trunc / JSON-extract SQL that runs across all six SQL dialects.

Honest ceiling: by ~10M events/day, time-range aggregates over 30 days take >5s on Postgres. The _voltro_events table is bounded by a retention sweep (VOLTRO_EVENTS_TTL_HOURS, default 365 days) so it never grows without limit; past ~10M events/day, swap to a real OLAP sink before queries get slow.

Tenant-scoped reads: useAnalytics() stamps the caller's subject.tenantId onto every track and every aggregate / timeseries / topN, so a handler can't write or read across tenants. A system/background context with no subject gets the raw (cross-tenant) sink; set an explicit tenantId on the event/query only for a deliberate cross-tenant op.

#@voltro/plugin-duckdb

Embedded DuckDB sidecar via @duckdb/node-api. Real column-store + vectorized execution, in-process — no external service.

import { duckdbAnalytics } from '@voltro/plugin-duckdb'

analytics: duckdbAnalytics({
  path: '.voltro/analytics.duckdb',   // or omit / ':memory:' for ephemeral
}),

DuckDB is the quiet win for ~80% of growth-stage apps: real OLAP performance with zero external service to deploy. The plugin owns its own voltro_events table inside the DuckDB instance and queries against it.

Scope:

• Events + opt-in CDC-mirror. track() writes to DuckDB; queries read from DuckDB. Pass mirrorTables to stream the main DataStore's reactive-table changes into voltro_mirror_<table> tables inside DuckDB so analytical queries can JOIN events against live user data (see CDC-mirror below). Without mirrorTables the sink is events-only.
• Single-process. DuckDB can't open the same file from multiple workers. For multi-instance deployments either pin analytics traffic to one replica or use clickhouse instead.

#@voltro/plugin-clickhouse

Production OLAP via the official @clickhouse/client. Self-hosted ClickHouse OR ClickHouse Cloud.

import { clickhouseAnalytics } from '@voltro/plugin-clickhouse'

analytics: clickhouseAnalytics({
  url:      process.env.CLICKHOUSE_URL!,
  database: 'voltro_events',
  username: process.env.CLICKHOUSE_USER,
  password: process.env.CLICKHOUSE_PASSWORD,
}),

The plugin owns the events table schema (MergeTree engine, ORDER BY (name, occurred_at, id), LowCardinality(String) on the event name, ZSTD-compressed properties JSON). First boot creates it; the plugin pings the cluster to fail-fast on bad config.

Opt-in batching. By default every track() is one immediate HTTP insert. Pass batch: { maxSize?, flushIntervalMs? } to buffer rows and flush them in ONE multi-row insert by size (default 1000), on a timer (default 5000 ms), and on shutdown (a graceful drain before the client closes). This trades per-event delivery confirmation for far fewer round-trips under load — with batching a successful track() means "buffered", and a later flush failure is logged + the batch dropped (best-effort), so leave batch unset when you need per-event delivery confirmation. Retention (a TTL on the events table) stays the operator's job.

There is no raw-client escape hatch: HyperLogLog, dictionaries, materialised views — the things you actually picked ClickHouse for — live outside the cross-provider contract, and the plugin exposes no handle to the raw @clickhouse/client. Where you need them, query ClickHouse with your own client instance against the same tables. Stay on useAnalytics() for code that should remain provider-portable.

#@voltro/plugin-tinybird

Hosted ClickHouse via Tinybird's Events API + Pipes.

import { tinybirdAnalytics } from '@voltro/plugin-tinybird'

analytics: tinybirdAnalytics({
  token:      process.env.TINYBIRD_TOKEN!,
  region:     'eu',                     // 'eu' | 'us-east' | 'us-west' | 'asia-southeast'
  datasource: 'voltro_events',
  // pipes: { aggregate: 'my_agg', timeseries: 'my_ts', topN: 'my_top' },
}),

The plugin expects three canonical pipes to exist in your workspace:

• events_aggregate — receives event, from, to, metric_expr, filter_sql
• events_timeseries — same + bucket_fn
• events_topn — same + group_expr, limit

We don't synthesize pipes from the framework — Tinybird's .pipe DSL is too rich to generate from a generic spec. Push the three canonical pipes once via the tb CLI; override names with pipes: { ... } if your team uses a different convention.

#@voltro/plugin-posthog

PostHog forwards track() only. Funnels / cohorts / sessions / feature flags live in PostHog's own UI and SQL — the plugin's aggregate / timeseries / topN return AnalyticsCapabilityNotSupported so callers fall back cleanly.

import { posthogAnalytics } from '@voltro/plugin-posthog'

analytics: posthogAnalytics({
  apiKey: process.env.POSTHOG_KEY!,
  host:   'https://eu.posthog.com',          // optional, default app.posthog.com
}),

Typical wiring is compose with a primary sink that handles reads (see below) — PostHog mirrors every event for product-analytics insights while your primary sink owns the typed aggregate queries.

#composeAnalytics — multi-sink

For dual-write (e.g. postgres-lite for typed reads + PostHog for product analytics):

import { composeAnalytics } from '@voltro/runtime'
import { postgresAnalytics } from '@voltro/plugin-analytics-postgres'
import { posthogAnalytics }  from '@voltro/plugin-posthog'

export default defineApi({
  name: 'myApi',
  store: 'postgres',
  analytics: composeAnalytics([
    postgresAnalytics(),                                   // primary — handles reads
    posthogAnalytics({ apiKey: process.env.POSTHOG_KEY! }), // mirrors every track()
  ]),
})

Semantics:

• track() fans out to every sink in parallel. Failures are isolated per sink — one provider returning HTTP 503 doesn't fail the postgres insert.
• aggregate / timeseries / topN query to the first sink that supports the op. PostHog doesn't support these → composition queries to the next sink (postgres-lite).
• If no sink supports a read op, the composite returns AnalyticsCapabilityNotSupported({ provider: 'compose' }).

#Capability-not-supported error handling

The cross-provider contract intentionally surfaces capability gaps on the typed error channel. Catch them when you want graceful fallback:

import { Effect } from 'effect'

const result = yield* analytics.aggregate({
  event:  'match_completed',
  metric: 'count',
  range:  { from: daysAgo(7) },
}).pipe(
  Effect.catchTag('AnalyticsCapabilityNotSupported', () => Effect.succeed(0)),
  Effect.catchTag('AnalyticsError', (err) => Effect.gen(function* () {
    yield* Effect.logWarning('analytics aggregate failed', { provider: err.provider, cause: err.cause })
    return 0
  })),
)

The two typed errors:

• AnalyticsCapabilityNotSupported — the sink doesn't implement the operation (e.g. PostHog returning this for aggregate).
• AnalyticsError — the underlying provider raised something (HTTP failure, query syntax, connection lost).

#CDC-mirror of reactive tables

By default a sink stores only the events you track() — the main DataStore's reactive tables aren't in the warehouse, so analytical queries can't JOIN events against user data. Opt in with mirrorTables: the framework subscribes to the store's change stream and upserts/deletes the changed rows into a corresponding warehouse table, idempotently (keyed on the primary key).

import { duckdbAnalytics } from '@voltro/plugin-duckdb'

analytics: duckdbAnalytics({
  path: '.voltro/analytics.duckdb',
  mirrorTables: ['users', 'teams'],   // reactive tables to mirror
  // mirrorPrimaryKey: 'id',          // default 'id'
}),

The postgres-lite (postgresAnalytics({ mirrorTables: [...] })) and ClickHouse (clickhouseAnalytics({ url, mirrorTables: [...] })) sinks take the same options. Each mirrored table lands as _voltro_mirror_<table> (voltro_mirror_<table> on DuckDB / ClickHouse) holding { id, data, version, is_deleted } — id is the source row's primary key, data is the full row as JSON, version orders the writes (see below) and is_deleted marks a tombstone. Analytical queries JOIN events against the mirror and filter tombstones out:

-- DuckDB: events per user tier
SELECT json_extract_string(m.data, '$.tier'), COUNT(*)
FROM voltro_events e
JOIN voltro_mirror_users m ON m.id = e.subject_id AND m.is_deleted = false
GROUP BY 1

The mirror is opt-in (omit mirrorTables → events-only) and idempotent — inserts/updates upsert by primary key, deletes write a tombstone by key, so a re-delivered change (e.g. after a reconnect) is a no-op-equivalent overwrite.

#Delivery guarantee

At-least-once for the lifetime of the process, ordered per row. Read that sentence literally — every clause is a promise the framework keeps, and the sentence stops where the implementation does:

• Retried, not dropped. A failing mirror write is retried with exponential backoff (retryAttempts, default 5). A write that outlives its retries is queued for repair: a timer re-reads the row's current state from your database and re-applies it, so the mirror converges on the truth rather than on a stale change that happened to be in flight.
• Ordered per row. Writes for the same primary key are applied one at a time, in commit order, and every write carries a version stamped when the change left the store — never a clock read inside the warehouse client. A change that arrives late therefore loses: ClickHouse's ReplacingMergeTree(version) keeps the highest version, and the DuckDB / postgres mirrors apply the update only when the incoming version is newer. Different rows are still mirrored concurrently.
• A delete is a tombstone, not a row removal. Filter is_deleted = false (is_deleted = 0 / FINAL on ClickHouse). A physical delete would leave nothing for a late, stale insert of the same key to lose against — the row would silently come back. The tombstone also keeps its version, which is what lets a replacement replica seed a key's numbering after a delete.
• N replicas converge to ONE row per change. Under postgres CDC / mysql binlog every replica observes every change and mirrors it, so a change is written N times — but the version is derived from the CHANGE (the warehouse's own high-water mark for the key, plus the change's position in the fleet stream), never from a replica's clock. The N duplicates are byte-identical, and the sink's version guard collapses them for free: no leader election, and no clock-skew window in which an older image could out-version a newer one. A replica that boots mid-stream reads each key's high-water mark from the warehouse before its first write, so it continues the fleet's numbering instead of restarting it.
• Never surfaces to the request. The OLTP write that produced the change has already committed; a warehouse outage is isolated into the log channel and the metrics below.
• A graceful shutdown is not a crash. On SIGINT / SIGTERM the mirror stops taking new changes and then settles what is already queued and in flight, before the sink itself is disposed. That drain is bounded (3 s of the teardown budget you set with VOLTRO_SHUTDOWN_GRACE_MS, default 10 s): a warehouse that has stopped answering cannot hold the process open until the orchestrator's SIGKILL, which would lose strictly more. The two outcomes log differently — analytics mirror drained at info, or a warn naming what was still pending when the deadline cut it, because that is the moment those counters can still be read.
• Not durable across a crash. The repair queue lives in memory. A change still awaiting repair when the process dies — SIGKILL, an OOM, a host failure — is lost, as is one evicted after repairQueueLimit. Both are logged at error level and counted by voltro_analytics_mirror_dropped_total — if that counter is non-zero, the affected tables need a re-seed.

Metrics: voltro_analytics_mirror_forwarded_total, ..._retries_total, ..._repair_queued_total, ..._dropped_total.

#Tuning

Every number the mirror picks on your behalf has a default and an environment override:

Env var	Default	Meaning
VOLTRO_ANALYTICS_MIRROR_RETRY_ATTEMPTS	5	Total attempts per mirror write (1 = no retry).
VOLTRO_ANALYTICS_MIRROR_RETRY_BASE_MS	100	First backoff delay; doubles per attempt.
VOLTRO_ANALYTICS_MIRROR_RETRY_MAX_MS	30000	Ceiling for the doubling backoff.
VOLTRO_ANALYTICS_MIRROR_REPAIR_INTERVAL_MS	60000	How often the repair loop re-drives exhausted changes (0 disables it).
VOLTRO_ANALYTICS_MIRROR_REPAIR_QUEUE_LIMIT	10000	Maximum keys held for repair before the oldest is dropped and counted.
VOLTRO_ANALYTICS_MIRROR_VERSION_STATE_LIMIT	100000	Maximum keys whose fleet-scope version state stays in memory; an evicted key re-seeds from the warehouse on its next change.

#Default — no sink configured

If app.config.ts doesn't set analytics, the framework provides a no-op sink. track() calls drop silently (logged once at boot so operators notice); aggregate / timeseries / topN fail with AnalyticsCapabilityNotSupported({ provider: 'noop' }). Apps without analytics setup never crash on useAnalytics().track(...).

#Anti-patterns

• Don't bypass useAnalytics() and call provider clients directly unless you specifically need a provider-locked feature. Going through the contract keeps handler code provider-portable.
• Don't compose two reading-capable sinks expecting both to be queried. Reads query to the first capable sink — that's by design (different sinks may hold different views of the data). If you genuinely need cross-sink reads, query each provider directly with its own client.
• Don't dump millions of properties per event. Sink storage is JSON; cardinality at the property level slows JSON-extract queries. If you have unbounded dimensions, restructure into a separate events table that the analytical query reads through joins.
• Don't expect aggregate queries to reflect new tracks instantly on every sink. By default ClickHouse and Tinybird ingest each track() as an immediate HTTP request, but the providers themselves make rows visible asynchronously (Tinybird's Events API acknowledges before the row is queryable). Two sinks add an opt-in client-side batch option (clickhouseAnalytics({ batch }), posthogAnalytics({ batch })) that buffers events and flushes them in one request by size / on a timer / on shutdown — a successful track() then means "buffered", not "delivered". Postgres-lite reflects writes immediately because they're synchronous to the main DB.

#Related

• The cross-cutting aggregate convention covers *.aggregate.ts files — pre-defined queries that materialise their result on a schedule. An aggregate's build function gets ctx.analytics (the configured sink) alongside ctx.store, so it can query the warehouse (ctx.analytics.topN(...)) for the "reduce 10B-row warehouse to a 100-row leaderboard" pattern. See the cross-plan section.
• *.subscribe.ts covers best-effort per-row reactive callbacks that fire after every commit. Different shape from track() (subscribe fires from CDC; track is explicit ingest).