Skip to content

feat: add PostgreSQL community analytics #125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
KingDoxik merged 2 commits into from
Aug 10, 2026
Merged

feat: add PostgreSQL community analytics #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
661c7c4
feat: add PostgreSQL community analytics
KingDoxik Aug 9, 2026
6c8b6eb
fix: separate community analytics surfaces
KingDoxik Aug 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 1 addition & 15 deletions .env.example
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,18 +33,6 @@ DATABASE_SSL=false
# DATABASE_PLATFORM_USERNAME=voidhash
# DATABASE_PLATFORM_PASSWORD=replace-with-a-random-password
# DATABASE_PLATFORM_SSL=false
# Optional analytics profile. Leave CLICKHOUSE_URL unset for the core stack.
# CLICKHOUSE_URL=http://clickhouse:8123
CLICKHOUSE_DATABASE=voidhash
CLICKHOUSE_ADMIN_USERNAME=voidhash_admin
CLICKHOUSE_ADMIN_PASSWORD=replace-with-a-random-password
CLICKHOUSE_USERNAME=voidhash_app
CLICKHOUSE_PASSWORD=replace-with-a-random-password
CLICKHOUSE_RO_USERNAME=voidhash_ro
CLICKHOUSE_RO_PASSWORD=replace-with-a-random-password
CLICKHOUSE_ANALYTICS_QUERY_USERNAME=voidhash_query
CLICKHOUSE_ANALYTICS_QUERY_PASSWORD=replace-with-a-random-password
CLICKHOUSE_HTTP_PORT=8123
MIMIC_ROOT_USERNAME=root
MIMIC_ROOT_PASSWORD=replace-with-a-random-password
PUBLIC_BASE_URL=http://localhost:5001
Expand Down Expand Up @@ -101,7 +89,7 @@ MAILPIT_UI_PORT=8025
# ── Local development & integration tests ────────────────────────────────────
# Used together with docker-compose.dev.yml:
# docker compose -f docker-compose.yml -f docker-compose.dev.yml \
# --profile analytics up -d --build
# up -d --build
# `pnpm test:integration` (repo root) reads this file and derives host-side
# connection settings from the values below, so the whole suite runs against
# this stack with no additional configuration.
Expand All @@ -111,8 +99,6 @@ MAILPIT_UI_PORT=8025
DATABASE_HOST_PORT=5432
COMPILER_HOST_PORT=5002

# To enable the analytics profile end-to-end (the compose service, migrations,
# and the ClickHouse integration suite), uncomment CLICKHOUSE_URL above.

# Browser used by the screenshot integration tests on the host. The container
# ships its own chromium; this is only for host-side test runs.
Expand Down
21 changes: 8 additions & 13 deletions .github/workflows/selfhost.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,20 +63,15 @@ jobs:
grep -E '^[A-Z][A-Z0-9_]*=' .env >> "$GITHUB_ENV"

- name: Start stateful stores
run: docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --env-file .env --profile analytics up -d clickhouse minio --wait --wait-timeout 180
run: docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --env-file .env up -d minio --wait --wait-timeout 180

- name: Initialize object store
run: docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --env-file .env --profile analytics run --rm minio-init
run: docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --env-file .env run --rm minio-init

# The dev overlay publishes Postgres and the compiler, which the
# host-side integration tier connects to. `CLICKHOUSE_URL` stays a shell
# override rather than an entry in `.env`: it names the compose-internal
# endpoint the application dials, while every host-side tier reaches
# ClickHouse on the published port instead.
# The dev overlay publishes PostgreSQL and the compiler, which the
# host-side integration tier connects to.
- name: Build and start Community Compose
env:
CLICKHOUSE_URL: http://clickhouse:8123
run: docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --env-file .env --profile analytics up --build --wait --wait-timeout 180
run: docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --env-file .env up --build --wait --wait-timeout 180

- name: Reclaim image build cache
run: docker builder prune --all --force
Expand All @@ -100,9 +95,9 @@ jobs:
- name: Show Compose diagnostics
if: always()
run: |
docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --profile analytics ps || true
docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --profile analytics logs --no-color || true
docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml ps || true
docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml logs --no-color || true

- name: Stop Compose
if: always()
run: docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml --profile analytics down --volumes --remove-orphans
run: docker compose -f selfhost/docker-compose.yml -f selfhost/docker-compose.dev.yml down --volumes --remove-orphans
1 change: 0 additions & 1 deletion apps/backend/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
"@voidhash/api-contracts": "workspace:*",
"@voidhash/app-store-server-sdk": "workspace:*",
"@voidhash/backend": "workspace:*",
"@voidhash/clickhouse-db": "workspace:*",
"@voidhash/core": "workspace:*",
"@voidhash/db": "workspace:*",
"@voidhash/lib": "workspace:*",
Expand Down
299 changes: 10 additions & 289 deletions apps/backend/src/backend/Analytics.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,300 +1,21 @@
import {
AnalyticsIngestQueueMessage,
type AnalyticsIngestQueueMessageType,
type AnalyticsWriterMessageType,
} from "@voidhash/core/domain/analyticsIngest/AnalyticsIngest";
import { IdentityProjectionPublisher } from "@voidhash/core/services";
import { AnalyticsIngestDlqService } from "@voidhash/core/services/analyticsIngest/AnalyticsIngestDlqService";
import { AnalyticsEventStore } from "@voidhash/core/services/analytics/AnalyticsEventStore";
import { AnalyticsDispatchService } from "@voidhash/core/services/analyticsIngest/AnalyticsDispatchService";
import { AnalyticsWriterService } from "@voidhash/core/services/analyticsIngest/AnalyticsWriterService";
import {
CaptureIngress,
CaptureIngressError,
type PublishableCaptureEvent,
} from "@voidhash/core/services/analyticsIngest/CaptureIngress";
import { DlqProducer } from "@voidhash/core/services/analyticsIngest/DlqProducer";
import { EventCaptureService } from "@voidhash/core/services/analyticsIngest/EventCaptureService";
import { EventProcessorService } from "@voidhash/core/services/analyticsIngest/EventProcessorService";
import {
PolicyCounterStore,
PolicyStoreError,
} from "@voidhash/core/services/analyticsIngest/PolicyCounterStore";
import { ClickhouseWebClient } from "@voidhash/clickhouse-db/clickhouse-client-web";
import { PersonIdentityService } from "@voidhash/core/services/personIdentity/PersonIdentityService";
import { Db } from "@voidhash/db";
import { KeyValueStore } from "@voidhash/platform/KeyValueStore";
import { PlatformRuntime } from "@voidhash/platform/PlatformRuntime";
import { QueueDriver } from "@voidhash/platform/Queue";
import { Context, Effect, Layer, Schema } from "effect";
import { Layer } from "effect";

import type { SelfhostRuntimeConfig } from "../config.ts";
import { makeSelfhostPlatformLive } from "./PlatformProfile.ts";

const analyticsQueueName = "analytics-ingest";
const analyticsDeadLetterQueueName = "analytics-ingest-dlq";

const minuteBucket = (value: Date): string => value.toISOString().slice(0, 16);
const dayBucket = (value: Date): string => value.toISOString().slice(0, 10);

const minuteMillis = 60_000;

/**
* Milliseconds left until the next UTC minute boundary. UTC minutes are aligned
* to the epoch, so this is exact modular arithmetic over the instant.
*/
const millisecondsUntilNextMinute = (value: Date): number => {
const remainder = value.getTime() % minuteMillis;
if (remainder === 0) return minuteMillis;
return minuteMillis - remainder;
};

/** JSON text of an ingest envelope as stored on the processed record. */
const encodeEnvelopeJson = Schema.encodeSync(Schema.UnknownFromJsonString);

/** Best-effort human text for an unknown queue/driver error. */
const errorCauseText = (error: unknown): string => {
if (typeof error === "object" && error !== null && "cause" in error) {
return String(error.cause);
}
return String(error);
};

const makePolicyCounterStoreLive = Layer.effect(
PolicyCounterStore,
Effect.gen(function* () {
const store = yield* KeyValueStore;
const runtime = yield* PlatformRuntime;
const increment = (key: string, ttlMillis: number) =>
store.increment("analytics-policy", key, { ttlMillis }).pipe(
Effect.provideService(PlatformRuntime, runtime),
Effect.mapError(
(error) =>
new PolicyStoreError({
cause: error.cause,
message: "policy counter increment failed",
}),
),
);
return PolicyCounterStore.of({
checkEventQuota: ({ now, projectId, quota }) => {
if (typeof quota !== "number" || quota < 1) return Effect.succeed(true);
return increment(`events:${projectId}:${dayBucket(now)}`, 172_800_000).pipe(
Effect.map((count) => count <= quota),
);
},
checkRequestLimit: ({ now, projectId, requestsPerMinute }) => {
if (typeof requestsPerMinute !== "number" || requestsPerMinute < 1) {
return Effect.succeed({ allowed: true });
}
return increment(`requests:${projectId}:${minuteBucket(now)}`, 120_000).pipe(
Effect.map((count) => {
if (count <= requestsPerMinute) return { allowed: true };
return { allowed: false, retryAfterMs: millisecondsUntilNextMinute(now) };
}),
);
},
});
}),
);

const makeCaptureIngressLive = Layer.effect(
CaptureIngress,
Effect.gen(function* () {
const queues = yield* QueueDriver;
const runtime = yield* PlatformRuntime;
const dlq = yield* AnalyticsIngestDlqService;
const producer = queues.producer(analyticsQueueName, AnalyticsIngestQueueMessage);

const enqueueBatch = (events: ReadonlyArray<PublishableCaptureEvent>) =>
Effect.gen(function* () {
const publishable: AnalyticsIngestQueueMessageType[] = [];
for (const event of events) {
if (
event.routeClass !== "main" &&
event.routeClass !== "overflow" &&
event.routeClass !== "historical"
) {
yield* dlq.recordFailure({
attemptCount: 0,
captureId: event.envelope.captureId,
distinctId: event.envelope.distinctId,
failureClass: "unsupported_route",
failureMessage: `route '${event.routeClass}' is not supported by queue ingest`,
payloadJson: event.envelope,
projectId: event.envelope.projectId,
routeClass: event.routeClass,
sourceSequence: 0,
sourceShard: "capture-ingress",
});
continue;
}
publishable.push({ envelope: event.envelope, lane: event.routeClass });
}
if (publishable.length > 0) {
yield* producer.publishBatch(publishable).pipe(
Effect.provideService(PlatformRuntime, runtime),
);
}
}).pipe(
Effect.mapError(
(error) =>
new CaptureIngressError({
cause: errorCauseText(error),
message: "failed to enqueue captured analytics events",
}),
),
);

return CaptureIngress.of({ enqueueBatch });
}),
);

/**
* Builds the process-wide platform primitives — queue, key-value store, cron
* scheduler, and runtime marker — plus the policy counter and capture services
* layered on top of them.
*/
export const makeSelfhostAnalyticsRuntimeLive = (config: SelfhostRuntimeConfig) => {
const platform = makeSelfhostPlatformLive(config);
/** Builds the synchronous PostgreSQL analytics services for Community self-host. */
export const makeSelfhostAnalyticsRuntimeLive = (
config: SelfhostRuntimeConfig,
): Layer.Layer<AnalyticsEventStore | AnalyticsDispatchService | EventCaptureService> => {
const database = Db.layer(config.database);
const dlq = AnalyticsIngestDlqService.layer.pipe(Layer.provide(database));
const policy = makePolicyCounterStoreLive.pipe(Layer.provide(platform));
const ingress = makeCaptureIngressLive.pipe(
Layer.provide(dlq),
Layer.provide(database),
Layer.provide(platform),
);
const store = AnalyticsEventStore.layer.pipe(Layer.provide(database));
const capture = EventCaptureService.layer.pipe(
Layer.provide(policy),
Layer.provide(ingress),
Layer.provide(store),
Layer.provide(database),
);
const dispatch = AnalyticsDispatchService.layer.pipe(Layer.provide(ingress));
return Layer.mergeAll(platform, capture, dispatch);
const dispatch = AnalyticsDispatchService.layer.pipe(Layer.provide(store));
return Layer.mergeAll(store, capture, dispatch);
};

/** Runs the analytics ingest and dead-letter consumers until their scope closes. */
export const runSelfhostAnalyticsConsumers = (
config: SelfhostRuntimeConfig,
clickhouse?: Layer.Layer<ClickhouseWebClient.ClickhouseWebClient>,
) =>
Effect.gen(function* () {
const queues = yield* QueueDriver;
const database = Db.layer(config.database);
const dlq = AnalyticsIngestDlqService.layer.pipe(Layer.provide(database));
const processor = EventProcessorService.layer.pipe(
Layer.provide(
DlqProducer.dbLive.pipe(
Layer.provide(dlq),
Layer.provide(database),
),
),
Layer.provide(
PersonIdentityService.layer.pipe(
Layer.provide(IdentityProjectionPublisher.noop),
Layer.provide(database),
),
),
Layer.provide(database),
);
const buildWriterContext = () => {
if (clickhouse === undefined) {
return Layer.build(AnalyticsWriterService.layer.pipe(Layer.provide(database)));
}
return Effect.gen(function* () {
const clickhouseContext = yield* Layer.build(clickhouse);
const client = Context.get(
clickhouseContext,
ClickhouseWebClient.ClickhouseWebClient,
);
return yield* Layer.build(
AnalyticsWriterService.layerWithClickhouse(client).pipe(
Layer.provide(database),
),
);
});
};
const writerContext = yield* buildWriterContext();
const analyticsWriter = Context.get(writerContext, AnalyticsWriterService);

const consumeAnalytics = queues.consumeBatch(
analyticsQueueName,
AnalyticsIngestQueueMessage,
(messages) =>
Effect.gen(function* () {
const eventProcessor = yield* EventProcessorService;
const writerMessages: AnalyticsWriterMessageType[] = [];
for (const message of messages) {
const outputs = yield* eventProcessor.processRecordToOutputs({
capturedEvent: message.envelope,
headers: {},
lane: message.lane,
rawValue: encodeEnvelopeJson(message.envelope),
sourceOffset: message.envelope.captureId,
sourcePartition: 0,
sourceTopic: message.envelope.routing.targetTopic,
});
for (const processed of outputs.processedEvents) {
writerMessages.push({
kind: "processed",
messageId: processed.processedEventId,
value: processed,
});
}
for (const person of outputs.personEvents) {
writerMessages.push({
kind: "person",
messageId: `${person.projectId}:${person.personId}:${person.version}`,
value: person,
});
}
for (const identity of outputs.personIdentityEvents) {
writerMessages.push({
kind: "person-distinct-id",
messageId: `${identity.projectId}:${identity.distinctId}:${identity.version}`,
value: identity,
});
}
}
if (writerMessages.length > 0) {
yield* analyticsWriter.writeMessages(writerMessages);
}
}).pipe(Effect.provide(processor)),
{
batchSize: 100,
deadLetterQueue: analyticsDeadLetterQueueName,
maxRetries: 10,
},
);

const consumeDeadLetters = queues.consumeBatch(
analyticsDeadLetterQueueName,
AnalyticsIngestQueueMessage,
(messages) =>
Effect.gen(function* () {
const service = yield* AnalyticsIngestDlqService;
yield* Effect.forEach(
messages,
(message) =>
service.recordFailure({
attemptCount: 10,
captureId: message.envelope.captureId,
distinctId: message.envelope.distinctId,
failureClass: "ingest_retry_exhausted",
failureMessage: "exhausted analytics ingest queue retries",
payloadJson: message.envelope,
projectId: message.envelope.projectId,
routeClass: message.envelope.routing.routeClass,
sourceSequence: 0,
sourceShard: analyticsDeadLetterQueueName,
}),
{ discard: true },
);
}).pipe(Effect.provide(dlq), Effect.provide(database)),
{ batchSize: 100, maxRetries: 5 },
);

return yield* Effect.all([consumeAnalytics, consumeDeadLetters], {
concurrency: "unbounded",
});
});
Loading
Loading