@yaebal/analytics

typed event tracking and funnels, straight from middleware: ctx.track("purchase", { amount: 9 }) fans out to whatever sinks you configure — posthog, plausible, your own sqlite/clickhouse table, a generic http collector, or a console log for local dev. event names and their properties are checked against a catalog you declare once, the same way @yaebal/feature-flags types flag keys — a typo'd event name or a missing required property is a compile error, not a silent gap in your funnel three months later.

install

pnpm add @yaebal/analytics

usage

install analytics() with bot.install(). it adds ctx.track(name, properties?) and ctx.identify(properties) to every handler's context — each call resolves userId/chatId from the current ctx before handing the event to your adapters. skip the events catalog entirely to keep ctx.track fully untyped (any string name, any properties).

import { createBot } from "yaebal";
import { analytics, consoleAdapter, p } from "@yaebal/analytics";

// swap consoleAdapter() for postHogAdapter / plausibleAdapter / sqliteAdapter / clickhouseAdapter
const bot = createBot(process.env.BOT_TOKEN!).install(
  analytics({
    // event names + properties are checked against this catalog — a typo'd name or a
    // missing required property is a compile error, not a silent gap in your funnel
    events: {
      start: true, // declared, untyped — any properties allowed
      purchase: { props: p.object({ amount: p.number() }) },
    },
    adapters: [consoleAdapter()],
  }),
);

bot.command("start", (ctx) => {
  ctx.track("start", { source: "deeplink" });
  return ctx.reply("hello!");
});

bot.command("buy", (ctx) => {
  ctx.track("purchase", { amount: 9 });
  return ctx.reply("thanks!");
});

bot.start();

the typed catalog

p is a tiny, zero-dependency runtime validator — p.string() / p.number() / p.boolean() / p.optional(field) / p.object({...}). it exists to give ctx.track both compile-time property checking and a runtime guard against a malformed call; reach for zod/valibot and hand-write a PropsSchema ({ parse(value) }) if you need more than that.

import { p } from "@yaebal/analytics";

const events = {
  // "true" — declared, but any properties allowed (no schema to enforce)
  onboarding_completed: true,
  // a schema — required/optional fields checked at both compile time and runtime
  purchase: {
    props: p.object({ amount: p.number(), currency: p.optional(p.string()) }),
    sample: 0.5,          // only half of these reach adapters — see "privacy & sampling"
    redact: ["email"],    // stripped from this event's properties specifically
    description: "a completed checkout",
  },
};

auto-capture

emit events for common update kinds with no manual ctx.track call: autoTrack: ["commands", "callback_queries", "messages"]. each kind emits a fixed event name with the dynamic bit as a property — command_used + { command: "start" }, callback_query + { data: "..." }, message_received + { contentType: "text" } — never an event name per command/callback payload, which would blow up your adapters' event schemas. a command message is only counted once even with both "commands" and "messages" enabled.

import { createBot } from "yaebal";
import { analytics, consoleAdapter } from "@yaebal/analytics";

// each kind emits a FIXED event name with the dynamic bit as a property —
// command_used + { command }, message_received + { contentType } — never an
// event name per command/message, which would blow up your adapters' schemas
const bot = createBot(process.env.BOT_TOKEN!).install(
  analytics({ adapters: [consoleAdapter()], autoTrack: ["commands", "messages"] }),
);

bot.command("start", (ctx) => ctx.reply("hi! no ctx.track() call here — autoTrack did it"));
bot.on("message:text", (ctx) => ctx.reply(`you said: ${ctx.text}`));

bot.start();

adapters

adapters take an already-constructed client and type it structurally, so this package depends on nothing and never dictates a driver version.

import { clickhouseAdapter, httpAdapter, plausibleAdapter, postHogAdapter, sqliteAdapter } from "@yaebal/analytics";
import { PostHog } from "posthog-node";
import { DatabaseSync } from "node:sqlite";

analytics({
  events,
  adapters: [
    postHogAdapter(new PostHog(process.env.POSTHOG_KEY!)),
    plausibleAdapter({ domain: "mybot.example" }),
    sqliteAdapter(new DatabaseSync("analytics.db")),
    clickhouseAdapter(clickhouseClient, { batchSize: 50 }),
    httpAdapter("https://collector.example/events", { headers: { authorization: "Bearer ..." } }),
  ],
  onError: (error, event) => console.error("analytics failed", event.name, error),
});

• postHogAdapter(client, options?) — client is anything shaped like posthog-node's PostHog (structural capture() + optional identify()/flush()). a custom distinctId() keeps userId reachable as a property, since it's no longer posthog's own identity.
• plausibleAdapter(options) — posts straight to plausible's events api, no client library. plausible dedupes "unique visitors" by hashing IP + user-agent; without a real per-user IP, every event from a bot process would look like the same visitor, so this adapter sends a deterministic synthetic X-Forwarded-For derived from userId/chatId instead. only scalar properties reach plausible — an object/array value is dropped rather than corrupting the breakdown.
• sqliteAdapter(db, options?) — db is node:sqlite's DatabaseSync, better-sqlite3, or anything with exec/prepare. creates its table (and a (name, created_at) index) on first use — see sqliteSchema() to run the DDL yourself. supports query(), so analyticsAdmin() works against it directly.
• clickhouseAdapter(client, options?) — client is anything shaped like @clickhouse/client (structural insert()). buffers and batch-inserts (batchSize, intervalMs, maxRetries, maxBuffered, onDrop) — a failed insert retries with backoff instead of losing the batch. clickhouse has no auto-migration like sqlite's db.exec, so run clickhouseSchema()'s DDL yourself once.
• httpAdapter(url, options?) — POST batches of events as JSON to any HTTP collector (umami, a mixpanel-compatible batch endpoint, your own). same batching/retry/backpressure as clickhouseAdapter.
• memoryAdapter() — an in-memory sink: for tests and small bots that want /stats without a database. supports query().
• consoleAdapter() — zero-config sink for local development.

import { clickhouseSchema } from "@yaebal/analytics";

await clickhouseClient.command({ query: clickhouseSchema() });

privacy & sampling

userId/chatId are personal data the moment they leave your process — a cloud posthog/plausible instance is a third party. analytics() has first-class controls instead of leaving this to you.

analytics({
  events,
  adapters: [postHogAdapter(posthog)],
  // hash userId/chatId before any adapter sees them — stable per id (funnels still group
  // correctly), not reversible by casual inspection, NOT a security control (small id spaces
  // are brute-forceable). pass a function instead (e.g. HMAC with a secret) for anything stronger.
  anonymize: "hash",
  // strip a property from every event, on top of any catalog entry's own `redact`
  redact: ["ip"],
  // consulted with the RAW (pre-anonymize) ids — an opt-out list keyed by real telegram ids
  // still works. a throwing/rejecting predicate fails OPEN, so a buggy check can't blackhole
  // every event.
  shouldTrack: (event) => !optedOut.has(event.userId),
  // load-shedding, not a stable per-user rollout — 10% of calls reach adapters, decided
  // fresh per call. a catalog entry's own `sample` overrides this for that event.
  sample: 0.1,
});

an in-chat admin surface

analyticsAdmin({ isAdmin, adapter }) — a telegram-native ops surface for events analytics() tracks, the same pattern as @yaebal/feature-flags's flagsAdmin. /analytics reports total events and top event names over the last 24h; /analytics 1h | 7d | 30d switch the window. pair it with memoryAdapter, sqliteAdapter, or clickhouseAdapter — posthogAdapter/plausibleAdapter don't implement query(), so use their own dashboards instead.

import { createBot } from "yaebal";
import { analytics, analyticsAdmin, memoryAdapter } from "@yaebal/analytics";

// memoryAdapter/sqliteAdapter/clickhouseAdapter all support query() — analyticsAdmin
// reads through it. posthogAdapter/plausibleAdapter don't (use their own dashboards).
const store = memoryAdapter();

const bot = createBot(process.env.BOT_TOKEN!)
  .install(analytics({ adapters: [store], autoTrack: ["commands"] }))
  .install(analyticsAdmin({ isAdmin: () => true, adapter: store }));

bot.command("start", (ctx) => ctx.reply("hi!"));

bot.start();

flushing on shutdown

buffered adapters (clickhouseAdapter, httpAdapter) hold events in memory between batches. analytics() wires flush() to bot.onStop automatically when installed on a Bot — bot.stop() won't resolve until the drain completes. on a bare Composer or a webhook/serverless deployment where bot.onStop never fires, call .flush() yourself.

const events = analytics({ adapters: [clickhouseAdapter(client)] });
bot.install(events); // flush() is now wired to bot.onStop automatically

// only needed off a real Bot (a bare Composer, or bot.onStop never firing in a
// webhook/serverless deployment):
someOtherShutdownHook(() => events.flush());

a unified collection point

plugins like @yaebal/broadcast already emit their own event stream (onEvent). build a standalone client with createAnalytics() and feed both ctx.track and other plugins' events into the same adapters with fromEvent.

import { analytics, createAnalytics, fromEvent } from "@yaebal/analytics";

const events = createAnalytics({ adapters: [postHogAdapter(posthog)] });

bot.install(analytics(events)); // ctx.track(...) now shares events's adapters

const jobs = new Broadcast({
  // ...
  onEvent: (event) => events.track(fromEvent("broadcast", event)),
});

api

AnalyticsControl interface

AnalyticsOptions

testing

memoryAdapter() replaces hand-rolling a { track: (e) => events.push(e) } fake in every test. drive the bot with @yaebal/test as usual.

import { createTestEnv } from "@yaebal/test";
import { analytics, memoryAdapter } from "@yaebal/analytics";

const store = memoryAdapter();

const bot = new Composer<Context>()
  .install(analytics({ adapters: [store] }))
  .command("start", (ctx) => {
    ctx.track("start");
    return ctx.reply("hi");
  });

await createTestEnv(bot).createUser().sendCommand("start");
// store.events => [{ name: "start", userId: ..., timestamp: ..., ... }]