Product analytics

1import { createAPI } from "yatta/api";2import { realtime } from "../func/realtime";3import { withSession } from "../func/middleware";4 5const route = createAPI("/events");6 7route.use(withSession);8 9route.post("/", async (ctx) => {10  const user = ctx.state.user as { id: string };11  const body = await ctx.json();12 13  // Accept and queue. Never await analytics inside a request handler.14  await jobs.enqueue("event:record", {15    userId: user.id,16    name: String(body.name).slice(0, 64),17    props: body.props ?? {},18    sessionId: ctx.cookies().sid ?? null,19  }, { priority: "low" });20 21  // Live counters via SSE, so an internal dashboard updates without polling.22  realtime.to("analytics:live").send("event.received", {23    name: body.name, at: Date.now(),24  });25 26  return new Response(null, { status: 202 });27});

1export const schema = {2  events: {3    id: col.uuid(),4    userId: col.text().nullable(),5    sessionId: col.text().nullable(),6    name: col.text(),7    props: col.json<Record<string, unknown>>().default({}),8    at: col.createdAt(),9  },10 11  // One row per (day, name, prop). Pre-aggregated, so a query touches12  // thousands of rows rather than millions.13  eventRollups: {14    day: col.text(),              // YYYY-MM-DD15    name: col.text(),16    prop: col.text(),             // "" for no prop17    value: col.text(),18    count: col.integer(),19  },20};

1cron.schedule("5 0 * * *", async () => {2  const yesterday = new Date(Date.now() - 86_400_000)3    .toISOString().slice(0, 10);4 5  // ON CONFLICT DO UPDATE makes this idempotent: re-running the rollup for a6  // day replaces it rather than double-counting.7  db.run(8    `INSERT INTO event_rollups (day, name, prop, value, count)9     SELECT ?, name, '', '', COUNT(*)10       FROM events11      WHERE substr(at, 1, 10) = ?12      GROUP BY name13     ON CONFLICT(day, name, prop, value)14       DO UPDATE SET count = count + excluded.count`,15    [yesterday, yesterday],16  );17 18  // Same shape for the one prop that matters most, here: the plan.19  db.run(20    `INSERT INTO event_rollups (day, name, prop, value, count)21     SELECT ?, name, 'plan',22            COALESCE(json_extract(props, '$.plan'), 'unknown'),23            COUNT(*)24       FROM events25      WHERE substr(at, 1, 10) = ? AND name = 'checkout.completed'26      GROUP BY name, value27     ON CONFLICT(day, name, prop, value)28       DO UPDATE SET count = count + excluded.count`,29    [yesterday, yesterday],30  );31 32  observer.log.info("Rolled up analytics", { day: yesterday });33});

1import { cache } from "../func/cache";2 3route.get("/summary", async (ctx) => {4  const days = Number(ctx.query().days ?? 30);5  const since = new Date(Date.now() - days * 86_400_000)6    .toISOString().slice(0, 10);7 8  return Response.json(9    await cache.remember(10      `analytics:summary:${days}`,11      async () => {12        const rows = db.run(13          `SELECT name, SUM(count) AS total14             FROM event_rollups15            WHERE day >= ? AND prop = ''16            GROUP BY name17            ORDER BY total DESC`,18          [since],19          "all",20        );21        return { days, events: rows };22      },23      // Safe to cache for a minute: a dashboard does not need second-level24      // freshness, and this keeps a popular route off the database.25      { ttl: "60s", tags: ["analytics"] },26    ),27  );28});

1const FUNNEL = ["signup", "onboarding", "first_project", "invite_teammate"];2 3route.get("/funnel", async (ctx) => {4  const days = Number(ctx.query().days ?? 30);5  const since = new Date(Date.now() - days * 86_400_000)6    .toISOString().slice(0, 10);7 8  const rows = db.run(9    `SELECT name, SUM(count) AS total10       FROM event_rollups11      WHERE day >= ? AND prop = '' AND name IN (${FUNNEL.map(() => "?").join(",")})12      GROUP BY name`,13    [since, ...FUNNEL],14    "all",15  );16 17  const counts = Object.fromEntries(rows.map((r) => [r.name, Number(r.total)]));18 19  // Each step is relative to the first, so the drop-off is readable directly.20  const top = counts[FUNNEL[0]] ?? 0;21 22  return Response.json({23    steps: FUNNEL.map((name, i) => ({24      name,25      count: counts[name] ?? 0,26      // Conversion from the previous step, which is what locates the leak.27      stepConversion: i === 028        ? 129        : (counts[name] ?? 0) / Math.max(counts[FUNNEL[i - 1]] ?? 0, 1),30      overallConversion: (counts[name] ?? 0) / Math.max(top, 1),31    })),32  });33});