Fas 1a: analytics tracker, ingestion endpoint, admin dashboards
This commit is contained in:
Sven (AAMOS AI)
@@ -0,0 +1,211 @@
|
||||
import type { FastifyInstance } from "fastify";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { z } from "zod";
|
||||
import { FUNNELS } from "@app/shared-types";
|
||||
import { errors } from "../lib/errors.js";
|
||||
|
||||
/**
|
||||
* Admin analytics dashboards: funnels and retention (spec §8.2).
|
||||
* Queries run against product_analytics_events; no materialized aggregates
|
||||
* required for beta volumes.
|
||||
*/
|
||||
export async function adminAnalyticsRoutes(app: FastifyInstance) {
|
||||
const admin = { preHandler: [app.requireAdmin] };
|
||||
|
||||
/** List available funnels. */
|
||||
app.get("/admin/v1/analytics/funnels", admin, async () => {
|
||||
return {
|
||||
funnels: Object.entries(FUNNELS).map(([name, steps]) => ({ name, steps })),
|
||||
};
|
||||
});
|
||||
|
||||
/** Compute a funnel between two UTC dates (YYYY-MM-DD). */
|
||||
app.get("/admin/v1/analytics/funnels/:name", admin, async (req) => {
|
||||
const params = z.object({ name: z.enum(Object.keys(FUNNELS) as [string, ...string[]]) }).parse(req.params);
|
||||
const query = z
|
||||
.object({
|
||||
startDate: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
|
||||
endDate: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
|
||||
})
|
||||
.parse(req.query);
|
||||
|
||||
const steps = FUNNELS[params.name as keyof typeof FUNNELS];
|
||||
const start = new Date(query.startDate);
|
||||
const end = new Date(query.endDate);
|
||||
end.setUTCDate(end.getUTCDate() + 1);
|
||||
|
||||
// Build a CTE chain: each step selects distinct users who completed the
|
||||
// current event after the previous step's timestamp.
|
||||
const ctes: string[] = [];
|
||||
const selects: string[] = [];
|
||||
for (let i = 0; i < steps.length; i++) {
|
||||
const prev = i > 0 ? `step${i - 1}` : null;
|
||||
const cteName = `step${i}`;
|
||||
const event = steps[i]!;
|
||||
if (i === 0) {
|
||||
ctes.push(
|
||||
`${cteName} AS (\n` +
|
||||
` SELECT DISTINCT user_id, household_id\n` +
|
||||
` FROM product_analytics_events\n` +
|
||||
` WHERE event_name = ${sqlParam(event)}\n` +
|
||||
` AND occurred_at >= ${sqlParam(start.toISOString())}\n` +
|
||||
` AND occurred_at < ${sqlParam(end.toISOString())}\n` +
|
||||
`)`,
|
||||
);
|
||||
} else {
|
||||
ctes.push(
|
||||
`${cteName} AS (\n` +
|
||||
` SELECT DISTINCT ${prev!}.user_id, ${prev!}.household_id\n` +
|
||||
` FROM ${prev!}\n` +
|
||||
` INNER JOIN product_analytics_events e ON e.user_id = ${prev!}.user_id\n` +
|
||||
` WHERE e.event_name = ${sqlParam(event)}\n` +
|
||||
` AND e.occurred_at >= ${sqlParam(start.toISOString())}\n` +
|
||||
` AND e.occurred_at < ${sqlParam(end.toISOString())}\n` +
|
||||
`)`,
|
||||
);
|
||||
}
|
||||
selects.push(`(SELECT count(*) FROM ${cteName}) AS step${i}`);
|
||||
}
|
||||
|
||||
const querySql = `WITH ${ctes.join(", ")} SELECT ${selects.join(", ")}`;
|
||||
const result = await app.db.execute(sql.raw(querySql));
|
||||
const row = (Array.isArray(result) ? result[0] : (result.rows[0] ?? {})) as Record<string, number>;
|
||||
|
||||
const stepCounts = steps.map((event, i) => ({
|
||||
step: i + 1,
|
||||
event,
|
||||
count: Number(row[`step${i}`] ?? 0),
|
||||
}));
|
||||
|
||||
return {
|
||||
name: params.name,
|
||||
startDate: query.startDate,
|
||||
endDate: query.endDate,
|
||||
steps: stepCounts,
|
||||
conversionFromFirst: stepCounts.map((s, i) => ({
|
||||
step: s.step,
|
||||
event: s.event,
|
||||
rate: i === 0 ? 1 : stepCounts[0]?.count ? s.count / stepCounts[0].count : 0,
|
||||
})),
|
||||
};
|
||||
});
|
||||
|
||||
/** Retention cohorts. Cohort = users with account_created on a given UTC day.
|
||||
* Returns active users per cohort day for 0..30 days after creation.
|
||||
*/
|
||||
app.get("/admin/v1/analytics/retention", admin, async (req) => {
|
||||
const query = z
|
||||
.object({
|
||||
startDate: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
|
||||
endDate: z.string().regex(/^\d{4}-\d{2}-\d{2}$/).optional(),
|
||||
})
|
||||
.parse(req.query);
|
||||
|
||||
const endDate = query.endDate ?? new Date().toISOString().slice(0, 10);
|
||||
const cohortStart = new Date(query.startDate);
|
||||
const cohortEnd = new Date(endDate);
|
||||
cohortEnd.setUTCDate(cohortEnd.getUTCDate() + 1);
|
||||
|
||||
const raw = `
|
||||
WITH cohort AS (
|
||||
SELECT DISTINCT user_id,
|
||||
(occurred_at AT TIME ZONE 'UTC')::date AS cohort_date
|
||||
FROM product_analytics_events
|
||||
WHERE event_name = 'account_created'
|
||||
AND occurred_at >= ${sqlParam(cohortStart.toISOString())}
|
||||
AND occurred_at < ${sqlParam(cohortEnd.toISOString())}
|
||||
),
|
||||
activity AS (
|
||||
SELECT user_id,
|
||||
(occurred_at AT TIME ZONE 'UTC')::date AS active_date
|
||||
FROM product_analytics_events
|
||||
WHERE occurred_at >= ${sqlParam(cohortStart.toISOString())}
|
||||
),
|
||||
sizes AS (
|
||||
SELECT cohort_date, count(DISTINCT user_id) AS cohort_size
|
||||
FROM cohort
|
||||
GROUP BY cohort_date
|
||||
)
|
||||
SELECT s.cohort_date,
|
||||
d.day,
|
||||
s.cohort_size,
|
||||
count(DISTINCT a.user_id) AS active_users
|
||||
FROM sizes s
|
||||
CROSS JOIN generate_series(0, 30) AS d(day)
|
||||
LEFT JOIN cohort c ON c.cohort_date = s.cohort_date
|
||||
LEFT JOIN activity a
|
||||
ON a.user_id = c.user_id
|
||||
AND a.active_date = s.cohort_date + d.day
|
||||
GROUP BY s.cohort_date, d.day, s.cohort_size
|
||||
ORDER BY s.cohort_date, d.day
|
||||
`;
|
||||
|
||||
const result = await app.db.execute(sql.raw(raw));
|
||||
const rows = Array.isArray(result) ? result : result.rows;
|
||||
|
||||
const cohorts: Record<
|
||||
string,
|
||||
{ cohortSize: number; retention: Array<{ day: number; active: number; rate: number }> }
|
||||
> = {};
|
||||
for (const r of rows as Array<{ cohort_date: string; day: number; cohort_size: number; active_users: number }>) {
|
||||
const key = String(r.cohort_date).slice(0, 10);
|
||||
if (!cohorts[key]) {
|
||||
cohorts[key] = { cohortSize: Number(r.cohort_size), retention: [] };
|
||||
}
|
||||
const size = Number(r.cohort_size);
|
||||
cohorts[key].retention.push({
|
||||
day: Number(r.day),
|
||||
active: Number(r.active_users),
|
||||
rate: size > 0 ? Number(r.active_users) / size : 0,
|
||||
});
|
||||
}
|
||||
|
||||
return { cohorts };
|
||||
});
|
||||
|
||||
/** Daily event counts for trend charts. */
|
||||
app.get("/admin/v1/analytics/event-counts", admin, async (req) => {
|
||||
const query = z
|
||||
.object({
|
||||
startDate: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
|
||||
endDate: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
|
||||
eventName: z.string().max(64).optional(),
|
||||
})
|
||||
.parse(req.query);
|
||||
|
||||
const start = new Date(query.startDate);
|
||||
const end = new Date(query.endDate);
|
||||
end.setUTCDate(end.getUTCDate() + 1);
|
||||
|
||||
let filter = "";
|
||||
if (query.eventName) {
|
||||
filter = `AND event_name = ${sqlParam(query.eventName)}`;
|
||||
}
|
||||
|
||||
const raw = `
|
||||
SELECT (occurred_at AT TIME ZONE 'UTC')::date AS date,
|
||||
event_name,
|
||||
count(*) AS total,
|
||||
count(DISTINCT session_id) AS unique_sessions,
|
||||
count(DISTINCT user_id) AS unique_users
|
||||
FROM product_analytics_events
|
||||
WHERE occurred_at >= ${sqlParam(start.toISOString())}
|
||||
AND occurred_at < ${sqlParam(end.toISOString())}
|
||||
${filter}
|
||||
GROUP BY date, event_name
|
||||
ORDER BY date, event_name
|
||||
`;
|
||||
|
||||
const result = await app.db.execute(sql.raw(raw));
|
||||
const rows = Array.isArray(result) ? result : result.rows;
|
||||
return { counts: rows };
|
||||
});
|
||||
}
|
||||
|
||||
/** Escape a string/date parameter for raw SQL by wrapping it in single quotes.
|
||||
* This is intentionally local-only SQL; dates and enums are validated above.
|
||||
*/
|
||||
function sqlParam(value: string | number): string {
|
||||
if (typeof value === "number") return String(value);
|
||||
return `'${String(value).replace(/'/g, "''")}'`;
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
import type { FastifyInstance, FastifyRequest } from "fastify";
|
||||
import { isAnalyticsOptedIn, schema } from "@app/database";
|
||||
import {
|
||||
analyticsBatchInputSchema,
|
||||
MAX_ANALYTICS_BATCH_SIZE,
|
||||
validateAnalyticsPropertySize,
|
||||
} from "@app/validation";
|
||||
import { parse, errors } from "../lib/errors.js";
|
||||
import { audit } from "../lib/helpers.js";
|
||||
|
||||
/** Max events per batch is also the per-minute rate-limit budget for this endpoint. */
|
||||
const ANALYTICS_RATE_LIMIT_MAX = MAX_ANALYTICS_BATCH_SIZE;
|
||||
|
||||
/**
|
||||
* Product analytics ingestion (spec §8.2).
|
||||
* Accepts both authenticated and anonymous events. Anonymous events must have
|
||||
* anonymousId and can be backfilled to userId later when the user logs in.
|
||||
*/
|
||||
export async function analyticsRoutes(app: FastifyInstance) {
|
||||
app.post(
|
||||
"/v1/analytics/events",
|
||||
{
|
||||
config: {
|
||||
rateLimit: {
|
||||
max: ANALYTICS_RATE_LIMIT_MAX,
|
||||
timeWindow: "1 minute",
|
||||
keyGenerator: (req) => `analytics:${req.ip}:${(req as unknown as { userId?: string }).userId ?? "anon"}`,
|
||||
},
|
||||
},
|
||||
},
|
||||
async (req, reply) => {
|
||||
const input = parse(analyticsBatchInputSchema, req.body, "analytics batch");
|
||||
validateAnalyticsPropertySize(input.events);
|
||||
|
||||
let userId: string | undefined;
|
||||
try {
|
||||
const payload = await req.jwtVerify<{ sub: string; role: string; type: string }>();
|
||||
if (payload.type === "access" && payload.sub) {
|
||||
userId = payload.sub;
|
||||
}
|
||||
} catch {
|
||||
// Anonymous event – fine as long as the batch carries anonymousId.
|
||||
}
|
||||
const isAuthenticated = !!userId;
|
||||
|
||||
if (isAuthenticated && userId) {
|
||||
const optedIn = await isAnalyticsOptedIn(app.db, userId);
|
||||
if (!optedIn) {
|
||||
throw errors.forbidden("Analytics collection disabled by user consent.");
|
||||
}
|
||||
}
|
||||
|
||||
const now = new Date();
|
||||
const rows = input.events.map((event, index) => {
|
||||
const occurredAt = event.occurredAt ? new Date(event.occurredAt) : now;
|
||||
if (Number.isNaN(occurredAt.getTime())) {
|
||||
throw errors.badRequest(`Invalid occurredAt for event ${index}`);
|
||||
}
|
||||
return {
|
||||
occurredAt,
|
||||
receivedAt: now,
|
||||
eventName: event.name,
|
||||
anonymousId: event.anonymousId ?? null,
|
||||
sessionId: event.sessionId ?? null,
|
||||
userId: isAuthenticated ? userId : null,
|
||||
householdId: event.householdId ?? null,
|
||||
appVersion: event.appVersion ?? null,
|
||||
platform: event.platform ?? null,
|
||||
locale: event.locale ?? null,
|
||||
experimentVariant: event.experimentVariant ?? null,
|
||||
properties: event.properties ?? {},
|
||||
};
|
||||
});
|
||||
|
||||
// Single insert for the whole batch. On conflict is impossible because
|
||||
// events are insert-only with random UUIDs.
|
||||
const result = await app.db.insert(schema.productAnalyticsEvents).values(rows);
|
||||
const accepted = result.rowCount ?? rows.length;
|
||||
|
||||
if (isAuthenticated) {
|
||||
await audit(app.db, {
|
||||
actorUserId: userId,
|
||||
action: "analytics.ingest",
|
||||
targetType: "analytics_batch",
|
||||
targetId: String(accepted),
|
||||
ip: req.ip,
|
||||
});
|
||||
}
|
||||
|
||||
return reply.code(202).send({ accepted, rejected: 0 });
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -21,6 +21,8 @@ import { budgetRoutes } from "./routes/budget.js";
|
||||
import { subscriptionRoutes } from "./routes/subscriptions.js";
|
||||
import { communityRoutes } from "./routes/community.js";
|
||||
import { adminRoutes } from "./routes/admin.js";
|
||||
import { adminAnalyticsRoutes } from "./routes/admin-analytics.js";
|
||||
import { analyticsRoutes } from "./routes/analytics.js";
|
||||
|
||||
declare module "fastify" {
|
||||
interface FastifyInstance {
|
||||
@@ -80,6 +82,8 @@ export async function buildServer(config: AppConfig) {
|
||||
await app.register(subscriptionRoutes);
|
||||
await app.register(communityRoutes);
|
||||
await app.register(adminRoutes);
|
||||
await app.register(adminAnalyticsRoutes);
|
||||
await app.register(analyticsRoutes);
|
||||
|
||||
return app;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user