Add GA4 and Amplitude business event analytics

.env.example

@@ -9,6 +9,13 @@ LOCAL_DATABASE_PATH=.data/postgres
 GOOGLE_CLIENT_ID=
 # Comma-separated Google-verified administrator emails; empty denies all admin access.
 ADMIN_EMAILS=
+# GA4 Measurement Protocol API secret stays server-side (Secret Manager in production).
+GA4_MEASUREMENT_ID=
+GA4_API_SECRET=
+# Amplitude ingestion key stays server-side; no query/admin secret is needed.
+AMPLITUDE_API_KEY=
+AMPLITUDE_REGION=US
+AMPLITUDE_PROJECT_URL=
 SESSION_SECRET=replace-with-at-least-32-random-characters
 RUNNER_TOKEN=replace-with-a-different-32-character-random-value
 LLM_PROVIDER=gemini

apps/server/src/admin.ts

@@ -71,7 +71,7 @@ export async function registerAdmin(app: FastifyInstance, db: DB) {
         (SELECT count(DISTINCT user_id)::int FROM visitor_days,clock WHERE day>=date_trunc('week',today)::date AND day<=today) AS weekly,
         (SELECT count(DISTINCT user_id)::int FROM visitor_days,clock WHERE day>=date_trunc('month',today)::date AND day<=today) AS monthly,
         (SELECT count(*)::int FROM users) AS users,
-        (SELECT count(*)::int FROM users WHERE created_at >= $1::date AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1) AT TIME ZONE 'Asia/Seoul') AS new_users,
+        (SELECT count(*)::int FROM users WHERE created_at >= $1::date::timestamp AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1)::timestamp AT TIME ZONE 'Asia/Seoul') AS new_users,
         (SELECT count(DISTINCT user_id)::int FROM visitor_days WHERE day BETWEEN $1::date AND $2::date) AS period_visitors,
         (SELECT count(*)::int FROM support_tickets WHERE status!='RESOLVED') AS open_inquiries,
         (SELECT min(first_seen) FROM visitor_days) AS tracking_since`,
@@ -86,7 +86,7 @@ export async function registerAdmin(app: FastifyInstance, db: DB) {
         SELECT date_trunc($3,day)::date AS bucket,count(DISTINCT user_id)::int AS visitors FROM visitor_days WHERE day BETWEEN $1::date AND $2::date GROUP BY 1
       ), joined AS (
         SELECT date_trunc($3,created_at AT TIME ZONE 'Asia/Seoul')::date AS bucket,count(*)::int AS new_users FROM users
-        WHERE created_at >= $1::date AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1) AT TIME ZONE 'Asia/Seoul' GROUP BY 1
+        WHERE created_at >= $1::date::timestamp AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1)::timestamp AT TIME ZONE 'Asia/Seoul' GROUP BY 1
       ) SELECT to_char(b.bucket,'YYYY-MM-DD') AS period,CASE WHEN (SELECT min(day) FROM visitor_days) IS NULL OR (b.bucket+('1 ' || $3)::interval)::date <= (SELECT min(day) FROM visitor_days) THEN NULL ELSE coalesce(v.visitors,0)::int END AS visitors,coalesce(j.new_users,0)::int AS new_users
       FROM buckets b LEFT JOIN visitors v USING(bucket) LEFT JOIN joined j USING(bucket) ORDER BY b.bucket`,
           [...args, range.grain],
@@ -100,7 +100,7 @@ export async function registerAdmin(app: FastifyInstance, db: DB) {
       ).rows;
       const revenue = (
         await tx.query(
-          `SELECT currency,sum(CASE WHEN kind='PAYMENT' THEN amount ELSE 0 END)::text AS gross,sum(CASE WHEN kind='REFUND' THEN amount ELSE 0 END)::text AS refunds,sum(CASE WHEN kind='PAYMENT' THEN amount ELSE -amount END)::text AS net FROM revenue_entries WHERE occurred_at >= $1::date AT TIME ZONE 'Asia/Seoul' AND occurred_at < ($2::date+1) AT TIME ZONE 'Asia/Seoul' GROUP BY currency`,
+          `SELECT currency,sum(CASE WHEN kind='PAYMENT' THEN amount ELSE 0 END)::text AS gross,sum(CASE WHEN kind='REFUND' THEN amount ELSE 0 END)::text AS refunds,sum(CASE WHEN kind='PAYMENT' THEN amount ELSE -amount END)::text AS net FROM revenue_entries WHERE occurred_at >= $1::date::timestamp AT TIME ZONE 'Asia/Seoul' AND occurred_at < ($2::date+1)::timestamp AT TIME ZONE 'Asia/Seoul' GROUP BY currency`,
           args,
         )
       ).rows;
@@ -112,19 +112,19 @@ export async function registerAdmin(app: FastifyInstance, db: DB) {
       ).rows;
       const revenueSeries = (
         await tx.query(
-          `SELECT to_char(date_trunc($3,occurred_at AT TIME ZONE 'Asia/Seoul'),'YYYY-MM-DD') AS period,currency,sum(CASE WHEN kind='PAYMENT' THEN amount ELSE -amount END)::text AS amount FROM revenue_entries WHERE occurred_at >= $1::date AT TIME ZONE 'Asia/Seoul' AND occurred_at < ($2::date+1) AT TIME ZONE 'Asia/Seoul' GROUP BY 1,2 ORDER BY 1,2`,
+          `SELECT to_char(date_trunc($3,occurred_at AT TIME ZONE 'Asia/Seoul'),'YYYY-MM-DD') AS period,currency,sum(CASE WHEN kind='PAYMENT' THEN amount ELSE -amount END)::text AS amount FROM revenue_entries WHERE occurred_at >= $1::date::timestamp AT TIME ZONE 'Asia/Seoul' AND occurred_at < ($2::date+1)::timestamp AT TIME ZONE 'Asia/Seoul' GROUP BY 1,2 ORDER BY 1,2`,
           [...args, range.grain],
         )
       ).rows;
       const usage = (
         await tx.query(
-          `SELECT provider,model,count(*)::int AS calls,count(input_tokens)::int AS measured_calls,sum(input_tokens)::text AS input_tokens,sum(output_tokens)::text AS output_tokens,sum(thinking_tokens)::text AS thinking_tokens FROM review_api_usage WHERE created_at >= $1::date AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1) AT TIME ZONE 'Asia/Seoul' GROUP BY provider,model`,
+          `SELECT provider,model,count(*)::int AS calls,count(input_tokens)::int AS measured_calls,sum(input_tokens)::text AS input_tokens,sum(output_tokens)::text AS output_tokens,sum(thinking_tokens)::text AS thinking_tokens FROM review_api_usage WHERE created_at >= $1::date::timestamp AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1)::timestamp AT TIME ZONE 'Asia/Seoul' GROUP BY provider,model`,
           args,
         )
       ).rows;
       const events = (
         await tx.query(
-          `SELECT type,count(*)::int AS count FROM business_events WHERE created_at >= $1::date AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1) AT TIME ZONE 'Asia/Seoul' GROUP BY type ORDER BY count DESC`,
+          `SELECT type,count(*)::int AS count FROM business_events WHERE created_at >= $1::date::timestamp AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1)::timestamp AT TIME ZONE 'Asia/Seoul' GROUP BY type ORDER BY count DESC`,
           args,
         )
       ).rows;
@@ -168,7 +168,7 @@ export async function registerAdmin(app: FastifyInstance, db: DB) {
       .parse(req.query);
     const rows = (
       await db.query(
-        `SELECT id,type,actor_id,entity_id,metadata,created_at FROM business_events WHERE created_at >= $1::date AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1) AT TIME ZONE 'Asia/Seoul' AND ($3::text IS NULL OR type=$3) ORDER BY created_at DESC,id LIMIT 51 OFFSET $4`,
+        `SELECT id,type,actor_id,entity_id,metadata,created_at FROM business_events WHERE created_at >= $1::date::timestamp AT TIME ZONE 'Asia/Seoul' AND created_at < ($2::date+1)::timestamp AT TIME ZONE 'Asia/Seoul' AND ($3::text IS NULL OR type=$3) ORDER BY created_at DESC,id LIMIT 51 OFFSET $4`,
         [range.from, range.to, type ?? null, (page - 1) * 50],
       )
     ).rows;

apps/server/src/amplitude.ts

@@ -0,0 +1,184 @@
+import { randomUUID } from "node:crypto";
+import type { DB } from "./db.js";
+import {
+  eventProperties,
+  ga4EventName,
+  safeEventProperties,
+  type BusinessEventType,
+} from "./event-catalog.js";
+export type AmplitudeConfig = { apiKey: string; region: "US" | "EU" };
+export function amplitudeConfig(
+  env = process.env,
+): AmplitudeConfig | undefined {
+  if (!env.AMPLITUDE_API_KEY) return undefined;
+  const region = env.AMPLITUDE_REGION || "US";
+  if (
+    !/^[a-zA-Z0-9_-]{16,256}$/.test(env.AMPLITUDE_API_KEY) ||
+    !["US", "EU"].includes(region)
+  )
+    throw Error("INVALID_AMPLITUDE_CONFIGURATION");
+  return { apiKey: env.AMPLITUDE_API_KEY, region: region as "US" | "EU" };
+}
+export function amplitudeEvent(row: Record<string, any>) {
+  if (!Object.hasOwn(eventProperties, row.type))
+    throw Error("UNKNOWN_EVENT_TYPE");
+  const context = row.analytics_context ?? {};
+  return {
+    event_type: ga4EventName(row.type),
+    insert_id: row.id,
+    // Stable per-account fallback; do not leak Google IDs, email or DB row IDs.
+    device_id:
+      context.clientId ?? "server." + (row.analytics_id ?? "algostep-system"),
+    ...(row.analytics_id ? { user_id: row.analytics_id } : {}),
+    time: new Date(row.created_at).getTime(),
+    session_id:
+      context.clientId &&
+      Number.isSafeInteger(context.sessionId) &&
+      context.sessionId > 0
+        ? context.sessionId * 1000
+        : -1,
+    event_properties: {
+      ...safeEventProperties(row.type as BusinessEventType, row.metadata),
+      event_source: context.clientId ? "web_context" : "server_unattributed",
+    },
+    // No IP, device fingerprint, raw URLs, user properties or session replay.
+  };
+}
+export async function deliverAmplitude(
+  db: DB,
+  config = amplitudeConfig(),
+  send: typeof fetch = fetch,
+  limit = 250,
+) {
+  if (!config) return { configured: false, processed: 0 };
+  await db.query(
+    "UPDATE amplitude_deliveries SET status='PENDING',lease_token=NULL,lease_until=NULL,last_code='LEASE_RECOVERED',updated_at=now() WHERE status='PROCESSING' AND lease_until<now()",
+  );
+  // Provider deduplication is seven days. Stop before that window closes;
+  // expired payloads stay visible to operators instead of risking duplicates.
+  await db.query(
+    "UPDATE amplitude_deliveries SET status='EXPIRED',last_code='RETRY_WINDOW_EXPIRED',updated_at=now() WHERE status='PENDING' AND first_attempt_at<now()-interval '6 days'",
+  );
+  let processed = 0;
+  const started = Date.now();
+  while (processed < limit && Date.now() - started < 20000) {
+    const token = randomUUID();
+    const rows = await db.tx(async (tx) => {
+      const rows = (
+        await tx.query(
+          "SELECT e.*,u.analytics_id::text AS analytics_id,d.payload,d.attempts FROM amplitude_deliveries d JOIN business_events e ON e.id=d.event_id LEFT JOIN users u ON u.id=e.actor_id WHERE d.status='PENDING' AND d.available_at<=now() ORDER BY d.available_at,d.event_id LIMIT $1 FOR UPDATE OF d SKIP LOCKED",
+          [Math.min(10, limit - processed)],
+        )
+      ).rows;
+      const valid: Record<string, any>[] = [];
+      for (const row of rows) {
+        try {
+          const payload = row.payload ?? amplitudeEvent(row);
+          await tx.query(
+            "UPDATE amplitude_deliveries SET status='PROCESSING',attempts=attempts+1,first_attempt_at=COALESCE(first_attempt_at,now()),payload=$2,lease_token=$3,lease_until=now()+interval '90 seconds',updated_at=now() WHERE event_id=$1",
+            [row.id, JSON.stringify(payload), token],
+          );
+          valid.push({ ...row, payload, attempts: row.attempts + 1 });
+        } catch (error) {
+          // Only schema validation is handled here; DB failure rolls back.
+          if (
+            (error as any).name !== "ZodError" &&
+            (error as Error).message !== "UNKNOWN_EVENT_TYPE"
+          )
+            throw error;
+          await tx.query(
+            "UPDATE amplitude_deliveries SET status='REJECTED',last_code='INVALID_PAYLOAD',updated_at=now() WHERE event_id=$1",
+            [row.id],
+          );
+        }
+      }
+      return { valid, claimed: rows.length };
+    });
+    if (!rows.claimed) break;
+    processed += rows.claimed;
+    if (!rows.valid.length) continue;
+    let status = "PENDING",
+      code = "TRANSPORT_ERROR";
+    let delay = Math.min(
+      3600,
+      60 * 2 ** Math.min(6, Math.max(...rows.valid.map((r) => r.attempts))),
+    );
+    try {
+      const response = await send(
+        config.region === "EU"
+          ? "https://api.eu.amplitude.com/2/httpapi"
+          : "https://api2.amplitude.com/2/httpapi",
+        {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({
+            api_key: config.apiKey,
+            events: rows.valid.map((r) => r.payload),
+          }),
+          redirect: "error",
+          signal: AbortSignal.timeout(5000),
+        },
+      );
+      if (response.status === 200) {
+        const body = (await response.json()) as any;
+        if (
+          body.code === 200 &&
+          Number.isInteger(body.events_ingested) &&
+          body.events_ingested >= 0 &&
+          body.events_ingested <= rows.valid.length
+        ) {
+          status = "INGESTED";
+          code =
+            body.events_ingested === rows.valid.length
+              ? "INGESTED"
+              : "INGESTED_OR_DEDUPLICATED";
+          delay = 0;
+        } else code = "INGESTION_UNCONFIRMED";
+      } else if (response.status === 429 || response.status >= 500)
+        code = "HTTP_" + response.status;
+      else {
+        status = "REJECTED";
+        code = "HTTP_" + response.status;
+        delay = 0;
+      }
+    } catch {
+      /* Never log provider errors, payloads or credentials. */
+    }
+    await db.query(
+      "UPDATE amplitude_deliveries SET status=$3,last_code=$4,available_at=now()+$5*interval '1 second',lease_token=NULL,lease_until=NULL,updated_at=now() WHERE event_id=ANY($1::uuid[]) AND lease_token=$2",
+      [rows.valid.map((r) => r.id), token, status, code, delay],
+    );
+  }
+  return { configured: true, processed };
+}
+export async function amplitudeStatus(db: DB) {
+  const config = amplitudeConfig();
+  let url: string | null = null;
+  try {
+    const u = new URL(process.env.AMPLITUDE_PROJECT_URL ?? "");
+    if (
+      u.protocol === "https:" &&
+      [
+        "app.amplitude.com",
+        "analytics.amplitude.com",
+        "app.eu.amplitude.com",
+        "analytics.eu.amplitude.com",
+      ].includes(u.hostname) &&
+      !u.username &&
+      !u.password &&
+      !u.search &&
+      !u.hash
+    )
+      url = u.href;
+  } catch {}
+  return {
+    configured: !!config,
+    region: config?.region ?? null,
+    url,
+    states: (
+      await db.query(
+        "SELECT status,last_code,count(*)::int AS count,min(available_at) AS oldest FROM amplitude_deliveries GROUP BY status,last_code ORDER BY status",
+      )
+    ).rows,
+  };
+}

apps/server/src/analytics-context.ts

@@ -0,0 +1,19 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+export type AnalyticsContext = { clientId?: string; sessionId?: number };
+export const analyticsContext = new AsyncLocalStorage<AnalyticsContext>();
+export function readAnalyticsContext(
+  headers: Record<string, unknown>,
+): AnalyticsContext {
+  const client = headers["x-analytics-client-id"],
+    session = headers["x-analytics-session-id"];
+  return {
+    ...(typeof client === "string" && /^\d{1,20}\.\d{1,20}$/.test(client)
+      ? { clientId: client }
+      : {}),
+    ...(typeof session === "string" &&
+    /^\d{1,12}$/.test(session) &&
+    Number(session) > 0
+      ? { sessionId: Number(session) }
+      : {}),
+  };
+}