diff --git a/.env.example b/.env.example index e7d185c..8b31b18 100644 --- a/.env.example +++ b/.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 diff --git a/apps/server/src/admin.ts b/apps/server/src/admin.ts index f752e1e..acd0ebc 100644 --- a/apps/server/src/admin.ts +++ b/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; diff --git a/apps/server/src/amplitude.ts b/apps/server/src/amplitude.ts new file mode 100644 index 0000000..eeecb1e --- /dev/null +++ b/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) { + 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 { + 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[] = []; + 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, + }; +} diff --git a/apps/server/src/analytics-context.ts b/apps/server/src/analytics-context.ts new file mode 100644 index 0000000..440d657 --- /dev/null +++ b/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(); +export function readAnalyticsContext( + headers: Record, +): 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) } + : {}), + }; +} diff --git a/apps/server/src/app.ts b/apps/server/src/app.ts index 83ea2a7..86461fe 100644 --- a/apps/server/src/app.ts +++ b/apps/server/src/app.ts @@ -25,6 +25,9 @@ import { externalize, hydrate, type ObjectStorage } from "./object-storage.js"; import { maintenance } from "./maintenance.js"; import { registerAdmin } from "./admin.js"; import { businessEvent, visit } from "./business.js"; +import { analyticsContext, readAnalyticsContext } from "./analytics-context.js"; +import { registerClientEvents } from "./client-events.js"; +import { ga4Config } from "./ga4.js"; type Config = { db: DB; @@ -93,6 +96,9 @@ export async function createApp(c: Config) { logger: process.env.NODE_ENV !== "test", bodyLimit: 2 * 1024 * 1024, }); + app.addHook("onRequest", (req, _reply, done) => + analyticsContext.run(readAnalyticsContext(req.headers), done), + ); await app.register(cookie); await app.register(cors, { origin: c.origin, credentials: true }); await app.register(rateLimit, { @@ -229,7 +235,7 @@ export async function createApp(c: Config) { ); if ( rows[0].client === "android" && - !/^\/api\/(me|topics|quiz\/[^/?]+|auth\/logout|analytics\/visit|support)(\?|$)/.test( + !/^\/api\/(me|topics|quiz\/[^/?]+|auth\/logout|analytics\/(visit|events)|support)(\?|$)/.test( req.url, ) ) @@ -249,6 +255,7 @@ export async function createApp(c: Config) { ); }); await registerAdmin(app, c.db); + await registerClientEvents(app, c.db); const user = (req: any) => req.user.id as string; app.get("/api/operations/metrics", async () => ({ reviewApiUsage30Days: ( @@ -305,6 +312,7 @@ export async function createApp(c: Config) { ); app.get("/api/health", async () => ({ ok: true })); app.get("/api/config", async () => ({ + ga4MeasurementId: ga4Config()?.measurementId ?? null, googleClientId: c.googleClientId, reviewEnabled: c.llmEnabled, paymentEnabled: false, @@ -389,7 +397,14 @@ export async function createApp(c: Config) { path: "/", maxAge: 7 * 86400, }); - return { user: { ...account, isAdmin: isAdmin(account) }, csrf }; + return { + user: { + ...account, + isAdmin: isAdmin(account), + analyticsId: account.analytics_id, + }, + csrf, + }; }, ); app.get("/api/me", async (req: any) => { @@ -400,14 +415,18 @@ export async function createApp(c: Config) { name: req.user.display_name, email: req.user.email, isAdmin: isAdmin(req.user), + analyticsId: req.user.analytics_id, }, csrf: req.user.csrf, }; }); app.post("/api/auth/logout", async (req: any, reply) => { - await c.db.query("DELETE FROM sessions WHERE id=$1", [ - sessionDigest(req.cookies.session), - ]); + await c.db.tx(async (db) => { + await db.query("DELETE FROM sessions WHERE id=$1", [ + sessionDigest(req.cookies.session), + ]); + await businessEvent(db, "LOGOUT", user(req), null, "logout:" + id()); + }); reply.clearCookie("session", { path: "/" }); return { ok: true }; }); @@ -557,17 +576,27 @@ export async function createApp(c: Config) { .extend({ revision: z.number().int().nonnegative() }) .parse(req.body); await ownRecord(c.db, user(req), uuid.parse(req.params.id)); - const { rows } = await c.db.query( - "UPDATE records SET source=$1,language=$2,revision=revision+1,updated_at=now() WHERE id=$3 AND user_id=$4 AND revision=$5 AND expires_at>now() RETURNING *", - [v.source, v.language, req.params.id, user(req), v.revision], - ); - if (!rows[0]) - throw new DomainError( - 409, - "SAVE_CONFLICT", - "다른 화면에서 코드가 변경됐어요. 현재 코드를 복사한 뒤 최신 기록을 다시 열어 주세요.", + return c.db.tx(async (db) => { + const { rows } = await db.query( + "UPDATE records SET source=$1,language=$2,revision=revision+1,updated_at=now() WHERE id=$3 AND user_id=$4 AND revision=$5 AND expires_at>now() RETURNING *", + [v.source, v.language, req.params.id, user(req), v.revision], + ); + if (!rows[0]) + throw new DomainError( + 409, + "SAVE_CONFLICT", + "다른 화면에서 코드가 변경됐어요. 현재 코드를 복사한 뒤 최신 기록을 다시 열어 주세요.", + ); + await businessEvent( + db, + "CODE_SAVED", + user(req), + rows[0].id, + "save:" + rows[0].id + ":" + rows[0].revision, + { language: rows[0].language, revision: rows[0].revision }, ); - return rows[0]; + return rows[0]; + }); }); app.delete("/api/records/:id", async (req: any) => { await c.db.tx(async (db) => { @@ -745,6 +774,13 @@ export async function createApp(c: Config) { execution: await hydrate(c.storage, execution), reviews, }); + await businessEvent( + c.db, + "PDF_GENERATED", + user(req), + record.id, + "pdf:" + id(), + ); return reply .type("application/pdf") .header( @@ -779,8 +815,8 @@ export async function createApp(c: Config) { if (busy >= fleet.slots) return { job: null }; } - await db.query( - "UPDATE executions SET status='FAILED',result=$1,finished_at=now(),lease_token=NULL WHERE status='RUNNING' AND lease_expires_at= 10) return { job: null }; const e = ( await db.query( - "SELECT e.* FROM executions e JOIN records r ON r.id=e.record_id WHERE e.status='QUEUED' AND r.expires_at>now() ORDER BY e.created_at LIMIT 1 FOR UPDATE OF e SKIP LOCKED", + "SELECT e.*,r.user_id FROM executions e JOIN records r ON r.id=e.record_id WHERE e.status='QUEUED' AND r.expires_at>now() ORDER BY e.created_at LIMIT 1 FOR UPDATE OF e SKIP LOCKED", ) ).rows[0]; if (!e) return { job: null }; @@ -810,6 +862,14 @@ export async function createApp(c: Config) { "UPDATE executions SET status='RUNNING',attempt=attempt+1,lease_token=$2,lease_expires_at=now()+interval '90 seconds',worker_name=$3 WHERE id=$1", [e.id, token, fleet ? worker : null], ); + await businessEvent( + db, + "EXECUTION_STARTED", + e.user_id, + e.id, + "execution-start:" + e.id, + { attempt: e.attempt + 1 }, + ); const info = ( await db.query( "SELECT s.language,s.source,p.tests FROM code_snapshots s JOIN records r ON r.id=s.record_id JOIN problems p ON p.id=r.problem_id WHERE s.id=$1", diff --git a/apps/server/src/business.ts b/apps/server/src/business.ts index ccdeb61..8ec761e 100644 --- a/apps/server/src/business.ts +++ b/apps/server/src/business.ts @@ -1,25 +1,49 @@ import type { DB } from "./db.js"; import { randomUUID } from "node:crypto"; +import { + safeEventProperties, + type BusinessEventType, +} from "./event-catalog.js"; +import { analyticsContext } from "./analytics-context.js"; // Only purpose-selected non-content metadata belongs here: never code, prompts, // tokens, passwords, request bodies or inquiry contents. export async function businessEvent( db: DB, - type: string, + type: BusinessEventType, actor: string | null, entity: string | null, key: string, metadata: Record = {}, ) { await db.query( - "INSERT INTO business_events(id,event_key,type,actor_id,entity_id,metadata) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT(event_key) DO NOTHING", - [randomUUID(), key, type, actor, entity, JSON.stringify(metadata)], + `WITH inserted AS ( + INSERT INTO business_events(id,event_key,type,actor_id,entity_id,metadata,analytics_context) + VALUES($1,$2,$3,$4,$5,$6,CASE WHEN $7::jsonb!='{}'::jsonb THEN $7::jsonb ELSE COALESCE((SELECT analytics_context FROM business_events WHERE actor_id=$4 AND entity_id=$5 AND analytics_context!='{}'::jsonb ORDER BY created_at LIMIT 1),'{}'::jsonb) END) + ON CONFLICT(event_key) DO NOTHING RETURNING id), ga4 AS ( + INSERT INTO analytics_deliveries(event_id) SELECT id FROM inserted RETURNING event_id) + INSERT INTO amplitude_deliveries(event_id) SELECT event_id FROM ga4`, + [ + randomUUID(), + key, + type, + actor, + entity, + JSON.stringify(safeEventProperties(type, metadata)), + JSON.stringify(analyticsContext.getStore() ?? {}), + ], ); } export async function visit(db: DB, user: string) { await db.query( - "INSERT INTO visitor_days(user_id,day) VALUES($1,(now() AT TIME ZONE 'Asia/Seoul')::date) ON CONFLICT DO NOTHING", - [user], + `WITH v AS ( + INSERT INTO visitor_days(user_id,day) VALUES($1,(now() AT TIME ZONE 'Asia/Seoul')::date) ON CONFLICT DO NOTHING RETURNING day + ), e AS ( + INSERT INTO business_events(id,event_key,type,actor_id,metadata,analytics_context) + SELECT $2,'visit:' || $1 || ':' || day::text,'DAILY_VISIT',$1,'{}'::jsonb,$3::jsonb FROM v ON CONFLICT(event_key) DO NOTHING RETURNING id + ), ga4 AS (INSERT INTO analytics_deliveries(event_id) SELECT id FROM e RETURNING event_id) + INSERT INTO amplitude_deliveries(event_id) SELECT event_id FROM ga4`, + [user, randomUUID(), JSON.stringify(analyticsContext.getStore() ?? {})], ); } diff --git a/apps/server/src/client-events.ts b/apps/server/src/client-events.ts new file mode 100644 index 0000000..962739d --- /dev/null +++ b/apps/server/src/client-events.ts @@ -0,0 +1,70 @@ +import type { FastifyInstance } from "fastify"; +import type { DB } from "./db.js"; +import { z } from "zod"; +import { businessEvent } from "./business.js"; +import { + clientEventTypes, + eventProperties, + safeEventProperties, +} from "./event-catalog.js"; +import { analyticsStatus } from "./ga4.js"; +import { DomainError } from "./domain.js"; +export async function registerClientEvents(app: FastifyInstance, db: DB) { + app.get("/api/admin/analytics", () => analyticsStatus(db)); + app.post( + "/api/analytics/events", + { config: { rateLimit: { max: 60, timeWindow: "1 minute" } } }, + async (req: any) => { + const { events } = z + .object({ + events: z + .array( + z + .object({ + requestKey: z.string().uuid(), + type: z.enum(clientEventTypes), + properties: z.record(z.string(), z.unknown()), + }) + .strict(), + ) + .min(1) + .max(25), + }) + .strict() + .parse(req.body); + // Validate the whole batch before writing anything and disallow spoofed + // server actions (revenue, execution result, signup, review consumption). + const parsed = events.map((e) => ({ + ...e, + properties: eventProperties[e.type].strict().parse(e.properties), + })); + await db.tx(async (tx) => { + for (const e of parsed) { + const properties = safeEventProperties(e.type, e.properties); + if ( + properties.topic && + !( + await tx.query("SELECT id FROM topics WHERE id=$1", [ + properties.topic, + ]) + ).rows.length + ) + throw new DomainError( + 400, + "UNKNOWN_ANALYTICS_TOPIC", + "학습 주제를 확인해 주세요.", + ); + await businessEvent( + tx, + e.type, + req.user.id, + null, + "client:" + req.user.id + ":" + e.requestKey, + properties, + ); + } + }); + return { accepted: events.length }; + }, + ); +} diff --git a/apps/server/src/content-admin.ts b/apps/server/src/content-admin.ts index 325f38c..f62cf9e 100644 --- a/apps/server/src/content-admin.ts +++ b/apps/server/src/content-admin.ts @@ -1,6 +1,8 @@ import { readFile } from "node:fs/promises"; import { z } from "zod"; import { database } from "./db.js"; +import { businessEvent } from "./business.js"; +import { randomUUID } from "node:crypto"; const short = z.string().min(1).max(200), code = z.string().min(1).max(65536); const languages = z.object({ python: code, cpp: code, java: code }); @@ -105,6 +107,14 @@ if (process.argv[1]?.replaceAll("\\", "/").match(/content-admin\.(ts|js)$/)) { }), ], ); + await businessEvent( + tx, + "CONTENT_PUBLISHED", + null, + null, + "publish:" + randomUUID(), + { topics: data.topics.length, problems: data.problems.length }, + ); }); console.log("Catalog published atomically."); } finally { diff --git a/apps/server/src/event-catalog.ts b/apps/server/src/event-catalog.ts new file mode 100644 index 0000000..0af0bb0 --- /dev/null +++ b/apps/server/src/event-catalog.ts @@ -0,0 +1,125 @@ +import { z } from "zod"; +// A closed vocabulary: free text, code, URLs, search terms and identity details +// are never analytical properties, even if a client tries to submit them. +const code = z.string().regex(/^[a-zA-Z0-9_-]{1,80}$/); +const small = z.number().int().min(0).max(1000000); +const language = z.enum(["python", "cpp", "java"]); +const screen = z.enum([ + "home", + "lesson", + "practice", + "challenges", + "problems", + "workspace", + "history", + "support", + "admin", +]); +export const eventProperties = { + USER_REGISTERED: z.object({ client: z.enum(["web", "android"]) }), + LOGIN: z.object({ client: z.enum(["web", "android"]) }), + LOGOUT: z.object({}), + DAILY_VISIT: z.object({}), + QUIZ_ANSWERED: z.object({ correct: z.boolean(), topic: code }), + PRACTICE_STARTED: z.object({ problem: code, language }), + CODE_SAVED: z.object({ language, revision: small }), + EXECUTION_REQUESTED: z.object({ + mode: z.enum(["judge", "custom"]), + language, + problem: code, + }), + EXECUTION_STARTED: z.object({ attempt: small }), + EXECUTION_FINISHED: z.object({ + status: z.enum(["FAILED", "SUCCEEDED"]), + verdict: code, + }), + REVIEW_REQUESTED: z.object({ language }), + REVIEW_STARTED: z.object({}), + REVIEW_SUCCEEDED: z.object({ + model: z.string().regex(/^[a-zA-Z0-9._/-]{1,80}$/), + }), + REVIEW_FAILED: z.object({ reason: code }), + RECORD_DELETED: z.object({}), + PDF_GENERATED: z.object({}), + INQUIRY_CREATED: z.object({ + category: z.enum(["question", "bug", "billing", "other"]), + }), + INQUIRY_UPDATED: z.object({ + status: z.enum(["OPEN", "IN_PROGRESS", "RESOLVED"]), + }), + COSTS_IMPORTED: z.object({ entries: small }), + CONTENT_PUBLISHED: z.object({ topics: small, problems: small }), + SCREEN_VIEWED: z.object({ screen, topic: code.optional() }), + EXAMPLE_SELECTED: z.object({ topic: code, index: small }), + EXAMPLE_APPLIED: z.object({ + topic: code, + count: small, + success: z.boolean(), + }), + VISUALIZATION_CONTROL: z.object({ + action: z.enum([ + "previous", + "next", + "first", + "last", + "play", + "pause", + "seek", + "layout", + "zoom_in", + "zoom_out", + "fit", + "reset", + "move_node", + ]), + step: small.optional(), + }), + QUIZ_VIEWED: z.object({ topic: code, index: small }), + QUIZ_RETRIED: z.object({ topic: code }), + LANGUAGE_SELECTED: z.object({ language }), + RESULT_TAB_VIEWED: z.object({ + tab: z.enum(["visual", "tests", "analysis", "review"]), + }), + TEST_SELECTED: z.object({ index: small }), + PDF_EXPORTED: z.object({ + destination: z.enum(["download", "drive"]), + success: z.boolean(), + }), + SEARCH_USED: z.object({ result_count: small }), + CATEGORY_SELECTED: z.object({ index: small }), + THEME_CHANGED: z.object({ theme: z.enum(["light", "dark"]) }), + CLIENT_ERROR: z.object({ code: code }), + ERROR_RECOVERY: z.object({ action: z.enum(["retry", "home", "reload"]) }), + ADMIN_SECTION_VIEWED: z.object({ + section: z.enum(["overview", "finance", "users", "support", "events"]), + }), +} as const; +export type BusinessEventType = keyof typeof eventProperties; +export const clientEventTypes = [ + "SCREEN_VIEWED", + "EXAMPLE_SELECTED", + "EXAMPLE_APPLIED", + "VISUALIZATION_CONTROL", + "QUIZ_VIEWED", + "QUIZ_RETRIED", + "LANGUAGE_SELECTED", + "RESULT_TAB_VIEWED", + "TEST_SELECTED", + "PDF_EXPORTED", + "SEARCH_USED", + "CATEGORY_SELECTED", + "THEME_CHANGED", + "CLIENT_ERROR", + "ERROR_RECOVERY", + "ADMIN_SECTION_VIEWED", +] as const; +export function safeEventProperties( + type: BusinessEventType, + properties: unknown, +) { + return eventProperties[type].parse(properties) as Record< + string, + string | number | boolean + >; +} +export const ga4EventName = (type: string) => "as_" + type.toLowerCase(); diff --git a/apps/server/src/ga4.ts b/apps/server/src/ga4.ts new file mode 100644 index 0000000..a05daab --- /dev/null +++ b/apps/server/src/ga4.ts @@ -0,0 +1,210 @@ +import type { DB } from "./db.js"; +import { createHash, randomUUID } from "node:crypto"; +import { + eventProperties, + ga4EventName, + safeEventProperties, + type BusinessEventType, +} from "./event-catalog.js"; + +export type Ga4Config = { measurementId: string; apiSecret: string }; +export function ga4Config(env = process.env): Ga4Config | undefined { + if (!env.GA4_MEASUREMENT_ID || !env.GA4_API_SECRET) return undefined; + if ( + !/^G-[A-Z0-9]{4,20}$/.test(env.GA4_MEASUREMENT_ID) || + env.GA4_API_SECRET.length > 256 || + /\s/.test(env.GA4_API_SECRET) + ) + throw new Error("INVALID_GA4_CONFIGURATION"); + return { + measurementId: env.GA4_MEASUREMENT_ID, + apiSecret: env.GA4_API_SECRET, + }; +} +export function ga4Payload(row: Record) { + if (!Object.hasOwn(eventProperties, row.type)) + throw new Error("UNKNOWN_EVENT_TYPE"); + const properties = safeEventProperties( + row.type as BusinessEventType, + row.metadata, + ); + const context = row.analytics_context ?? {}; + // GA4 requires number.number even for server-only events. Hash the opaque + // analytics identity into that format, without inventing a browser session. + const fallback = createHash("sha256") + .update("algostep-ga4:" + (row.analytics_id ?? "system")) + .digest(); + const params: Record = { + event_id: row.id, + event_source: context.clientId ? "web_context" : "server_unattributed", + }; + for (const [key, value] of Object.entries(properties)) + params[key] = typeof value === "boolean" ? (value ? 1 : 0) : value; + if ( + context.clientId && + Number.isSafeInteger(context.sessionId) && + context.sessionId > 0 + ) + params.session_id = context.sessionId; + return { + client_id: + context.clientId ?? + `${fallback.readUInt32BE(0)}.${fallback.readUInt32BE(4)}`, + ...(row.analytics_id ? { user_id: row.analytics_id } : {}), + timestamp_micros: new Date(row.created_at).getTime() * 1000, + consent: { ad_user_data: "DENIED", ad_personalization: "DENIED" }, + validation_behavior: "ENFORCE_RECOMMENDATIONS", + events: [ + { + name: ga4EventName(row.type), + timestamp_micros: new Date(row.created_at).getTime() * 1000, + params, + }, + ], + }; +} + +// A 2xx collect response means accepted at the HTTP layer, not proven ingestion. +// Google provides no general idempotency contract: ambiguous collect responses +// are quarantined, never automatically resent. Operational metrics use the exact DB ledger. +export async function deliverGa4( + db: DB, + config = ga4Config(), + send: typeof fetch = fetch, + limit = 250, +) { + if (!config) return { configured: false, processed: 0 }; + const started = Date.now(); + let processed = 0; + await db.query( + "UPDATE analytics_deliveries SET status='UNCERTAIN',last_code='LEASE_EXPIRED',lease_token=NULL,lease_until=NULL,updated_at=now() WHERE status='PROCESSING' AND lease_until { + const next = ( + await tx.query( + "SELECT d.event_id,e.actor_id,e.analytics_context FROM analytics_deliveries d JOIN business_events e ON e.id=d.event_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", + ) + ).rows[0]; + if (!next) return []; + // One request has one client/user context. Preserve each event's original + // timestamp when batching up to the provider's 25-event limit. + const batch = ( + await tx.query( + "SELECT d.event_id FROM analytics_deliveries d JOIN business_events e ON e.id=d.event_id WHERE d.status='PENDING' AND d.available_at<=now() AND e.actor_id IS NOT DISTINCT FROM $1::uuid AND e.analytics_context=$2::jsonb ORDER BY d.available_at,d.event_id LIMIT $3 FOR UPDATE OF d SKIP LOCKED", + [ + next.actor_id, + JSON.stringify(next.analytics_context), + Math.min(25, limit - processed), + ], + ) + ).rows; + const ids = batch.map((r) => r.event_id); + await tx.query( + "UPDATE analytics_deliveries SET status='PROCESSING',lease_token=$2,lease_until=now()+interval '90 seconds',attempts=attempts+1,updated_at=now() WHERE event_id=ANY($1::uuid[])", + [ids, token], + ); + return ( + await tx.query( + "SELECT e.*,u.analytics_id::text AS analytics_id,d.attempts FROM business_events e LEFT JOIN users u ON u.id=e.actor_id JOIN analytics_deliveries d ON d.event_id=e.id WHERE e.id=ANY($1::uuid[])", + [ids], + ) + ).rows; + }); + if (!rows.length) break; + processed += rows.length; + let status = "REJECTED", + code = "INVALID_PAYLOAD", + collectStarted = false, + validationStarted = false, + delay = 0; + try { + const payload = ga4Payload(rows[0]); + payload.events = rows.flatMap((row) => ga4Payload(row).events); + const query = new URLSearchParams({ + measurement_id: config.measurementId, + api_secret: config.apiSecret, + }); + const options = { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(payload), + redirect: "error" as const, + }; + validationStarted = true; + const checked = await send( + "https://www.google-analytics.com/debug/mp/collect?" + query, + { ...options, signal: AbortSignal.timeout(5000) }, + ); + if (!checked.ok) { + code = "VALIDATION_HTTP_" + checked.status; + } else { + const result = (await checked.json()) as any; + if (!Array.isArray(result.validationMessages)) { + code = "INVALID_VALIDATION_RESPONSE"; + } else if (result.validationMessages.length) { + code = "VALIDATION_REJECTED"; + } else { + collectStarted = true; + const response = await send( + "https://www.google-analytics.com/mp/collect?" + query, + { ...options, signal: AbortSignal.timeout(5000) }, + ); + status = response.ok ? "ACCEPTED" : "REJECTED"; + code = response.ok + ? "HTTP_ACCEPTED" + : "COLLECT_HTTP_" + response.status; + } + } + } catch { + // Never log provider exceptions: their message may contain the secret URL. + if (collectStarted) { + status = "UNCERTAIN"; + code = "COLLECT_RESPONSE_UNKNOWN"; + } else if ( + validationStarted && + Math.max(...rows.map((r) => r.attempts)) < 5 + ) { + status = "PENDING"; + code = "VALIDATION_UNAVAILABLE"; + delay = Math.min( + 3600, + 60 * 2 ** Math.max(...rows.map((r) => r.attempts)), + ); + } else if (validationStarted) { + status = "REJECTED"; + code = "VALIDATION_RETRY_EXHAUSTED"; + } + } + await db.query( + "UPDATE analytics_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.map((r) => r.id), token, status, code, delay], + ); + } + return { configured: true, processed }; +} + +export async function analyticsStatus(db: DB) { + const config = ga4Config(); + const { amplitudeStatus } = await import("./amplitude.js"); + return { + ga4: { + configured: !!config, + measurementId: config?.measurementId ?? null, + states: ( + await db.query( + "SELECT status,last_code,count(*)::int AS count,min(available_at) AS oldest FROM analytics_deliveries GROUP BY status,last_code ORDER BY status", + ) + ).rows, + }, + amplitude: await amplitudeStatus(db), + catalog: Object.entries(eventProperties).map(([type]) => ({ + type, + ga4Name: ga4EventName(type), + })), + }; +} diff --git a/apps/server/src/maintenance.ts b/apps/server/src/maintenance.ts index 59efa60..f985e48 100644 --- a/apps/server/src/maintenance.ts +++ b/apps/server/src/maintenance.ts @@ -4,6 +4,8 @@ import type { DB } from "./db.js"; import { cleanExpired } from "./domain.js"; import { processReview, recoverReviews } from "./review-worker.js"; import { deleteObjects, type ObjectStorage } from "./object-storage.js"; +import { deliverGa4 } from "./ga4.js"; +import { deliverAmplitude } from "./amplitude.js"; export async function maintenance(db: DB, storage?: ObjectStorage) { await cleanExpired(db); await recoverReviews(db); @@ -22,5 +24,12 @@ export async function maintenance(db: DB, storage?: ObjectStorage) { } const llm = llmOptions(); if (llm) await processReview(db, llm); + // Independent destinations: one provider failure must not starve the other. + const delivered = await Promise.allSettled([ + deliverGa4(db), + deliverAmplitude(db), + ]); + if (delivered.some((r) => r.status === "rejected")) + throw new Error("ANALYTICS_MAINTENANCE_FAILED"); if (fleetError) throw fleetError; } diff --git a/apps/server/src/migrations.ts b/apps/server/src/migrations.ts index 284a861..f4176fa 100644 --- a/apps/server/src/migrations.ts +++ b/apps/server/src/migrations.ts @@ -46,6 +46,31 @@ const migrations = [ CREATE INDEX revenue_entries_date ON revenue_entries(occurred_at); `, }, + { + version: 6, + sql: ` + ALTER TABLE users ADD COLUMN analytics_id uuid NOT NULL DEFAULT gen_random_uuid(); + CREATE UNIQUE INDEX users_analytics_id ON users(analytics_id); + ALTER TABLE business_events ADD COLUMN analytics_context jsonb NOT NULL DEFAULT '{}'; + CREATE TABLE analytics_deliveries(event_id uuid PRIMARY KEY REFERENCES business_events ON DELETE CASCADE,status text NOT NULL DEFAULT 'PENDING' CHECK(status IN ('PENDING','PROCESSING','ACCEPTED','REJECTED','UNCERTAIN','EXPIRED')),attempts integer NOT NULL DEFAULT 0,available_at timestamptz NOT NULL DEFAULT now(),lease_token uuid,lease_until timestamptz,last_code text,updated_at timestamptz NOT NULL DEFAULT now()); + CREATE INDEX analytics_deliveries_queue ON analytics_deliveries(status,available_at); + INSERT INTO analytics_deliveries(event_id) SELECT id FROM business_events; + CREATE VIEW analytics_events WITH (security_barrier=true) AS SELECT e.id AS event_id,e.type AS event_type,'as_' || lower(e.type) AS ga4_event_name,u.analytics_id::text AS user_id,e.created_at,e.created_at AT TIME ZONE 'Asia/Seoul' AS occurred_at_kst,e.metadata FROM business_events e LEFT JOIN users u ON u.id=e.actor_id; + CREATE VIEW analytics_visitors WITH (security_barrier=true) AS SELECT u.analytics_id::text AS user_id,v.day AS day_kst FROM visitor_days v JOIN users u ON u.id=v.user_id; + CREATE VIEW analytics_users WITH (security_barrier=true) AS SELECT analytics_id::text AS user_id,created_at,created_at AT TIME ZONE 'Asia/Seoul' AS registered_at_kst FROM users; + CREATE VIEW analytics_costs WITH (security_barrier=true) AS SELECT usage_date AS day_kst,service,currency,sum(amount) AS amount FROM cost_entries GROUP BY usage_date,service,currency; + CREATE VIEW analytics_revenue WITH (security_barrier=true) AS SELECT (occurred_at AT TIME ZONE 'Asia/Seoul')::date AS day_kst,currency,kind,sum(amount) AS amount FROM revenue_entries GROUP BY 1,2,3; + CREATE VIEW analytics_inquiries WITH (security_barrier=true) AS SELECT (created_at AT TIME ZONE 'Asia/Seoul')::date AS day_kst,category,status,count(*) AS inquiries FROM support_tickets GROUP BY 1,2,3; + CREATE VIEW analytics_delivery_status WITH (security_barrier=true) AS SELECT e.type AS event_type,d.status,d.last_code,count(*) AS events,min(e.created_at) AS oldest_event,max(d.updated_at) AS last_update FROM analytics_deliveries d JOIN business_events e ON e.id=d.event_id GROUP BY 1,2,3; + REVOKE ALL ON analytics_events,analytics_visitors,analytics_users,analytics_costs,analytics_revenue,analytics_inquiries,analytics_delivery_status FROM PUBLIC; + `, + }, + { + version: 7, + sql: `CREATE TABLE amplitude_deliveries(event_id uuid PRIMARY KEY REFERENCES business_events ON DELETE CASCADE,status text NOT NULL DEFAULT 'PENDING' CHECK(status IN ('PENDING','PROCESSING','INGESTED','REJECTED','EXPIRED')),attempts integer NOT NULL DEFAULT 0,available_at timestamptz NOT NULL DEFAULT now(),first_attempt_at timestamptz,lease_token uuid,lease_until timestamptz,payload jsonb,last_code text,updated_at timestamptz NOT NULL DEFAULT now()); + CREATE INDEX amplitude_deliveries_queue ON amplitude_deliveries(status,available_at); + INSERT INTO amplitude_deliveries(event_id) SELECT id FROM business_events;`, + }, ]; export async function migrate(db: DB) { diff --git a/apps/server/src/review-worker.ts b/apps/server/src/review-worker.ts index e570048..b8cd4c8 100644 --- a/apps/server/src/review-worker.ts +++ b/apps/server/src/review-worker.ts @@ -20,6 +20,13 @@ export async function processReview(db: DB, options: LlmOptions) { "UPDATE reviews SET status='RUNNING',started_at=now() WHERE id=$1", [r.id], ); + await businessEvent( + tx, + "REVIEW_STARTED", + r.user_id, + r.id, + "review-start:" + r.id, + ); return r; }); if (!task) return false; diff --git a/apps/server/test/admin.test.ts b/apps/server/test/admin.test.ts index 297a90d..be43a96 100644 --- a/apps/server/test/admin.test.ts +++ b/apps/server/test/admin.test.ts @@ -310,8 +310,8 @@ test("cost import is atomic and repeatable; currencies and refunds stay separate }); assert.equal(operational.statusCode, 200); await db.query( - "INSERT INTO revenue_entries(id,provider_event_id,kind,currency,amount,occurred_at) VALUES($1,'payment-001','PAYMENT','KRW',3000,now()),($2,'refund-001','REFUND','KRW',1000,now())", - [id(), id()], + "INSERT INTO revenue_entries(id,provider_event_id,kind,currency,amount,occurred_at) VALUES($1,'payment-001','PAYMENT','KRW',3000,$3::date::timestamp AT TIME ZONE 'Asia/Seoul'),($2,'refund-001','REFUND','KRW',1000,($3::date::timestamp+interval '1 minute') AT TIME ZONE 'Asia/Seoul'),($4,'previous-day','PAYMENT','KRW',9999,($3::date::timestamp-interval '1 second') AT TIME ZONE 'Asia/Seoul'),($5,'next-day','PAYMENT','KRW',9999,($3::date+1)::timestamp AT TIME ZONE 'Asia/Seoul')", + [id(), id(), day, id(), id()], ); const r = await request( admin, @@ -374,7 +374,7 @@ test("business events remain after record deletion and retry keys do not inflate ); await assert.rejects( db.tx(async (tx) => { - await businessEvent(tx, "ROLLBACK_EVENT", student.id, null, "rollback"); + await businessEvent(tx, "LOGOUT", student.id, null, "rollback"); throw new Error("rollback"); }), ); diff --git a/apps/server/test/amplitude.test.ts b/apps/server/test/amplitude.test.ts new file mode 100644 index 0000000..ee76651 --- /dev/null +++ b/apps/server/test/amplitude.test.ts @@ -0,0 +1,133 @@ +import { test, before, after } from "node:test"; +import assert from "node:assert/strict"; +import { database, type DB } from "../src/db.js"; +import { businessEvent } from "../src/business.js"; +import { + amplitudeConfig, + amplitudeEvent, + deliverAmplitude, + amplitudeStatus, +} from "../src/amplitude.js"; +let db: DB; +const config = { apiKey: "test-amplitude-key", region: "US" as const }; +before(async () => { + db = await database(); +}); +after(async () => { + await db.close(); +}); +const reset = () => db.query("DELETE FROM business_events"); +const add = (key: string) => businessEvent(db, "LOGOUT", null, null, key); +const states = async () => + (await db.query("SELECT * FROM amplitude_deliveries ORDER BY event_id")).rows; +const success: typeof fetch = async (_url, options) => + new Response( + JSON.stringify({ + code: 200, + events_ingested: JSON.parse(String(options?.body)).events.length, + }), + ); +test("both destinations are atomically deduplicated and rollback together", async () => { + await reset(); + await add("same"); + await add("same"); + assert.equal((await states()).length, 1); + assert.equal( + (await db.query("SELECT * FROM analytics_deliveries")).rows.length, + 1, + ); + await assert.rejects( + db.tx(async (tx) => { + await businessEvent(tx, "LOGOUT", null, null, "rollback"); + throw Error("rollback"); + }), + ); + assert.equal((await states()).length, 1); +}); +test("Amplitude batches at most ten and uses fixed regional hosts without exposing secrets", async () => { + await reset(); + for (let i = 0; i < 12; i++) await add("batch-" + i); + const sizes: number[] = []; + await deliverAmplitude( + db, + { ...config, region: "EU" }, + async (url, options) => { + assert.equal(url, "https://api.eu.amplitude.com/2/httpapi"); + const body = JSON.parse(String(options?.body)); + sizes.push(body.events.length); + assert.equal(body.api_key, config.apiKey); + assert.equal(body.events[0].session_id, -1); + return success(url, options); + }, + ); + assert.deepEqual(sizes, [10, 2]); + assert.ok((await states()).every((r) => r.status === "INGESTED")); + assert.doesNotMatch( + JSON.stringify(await amplitudeStatus(db)), + /test-amplitude-key/, + ); +}); +test("ambiguous transport retries preserve payload and insert ID even after source changes", async () => { + await reset(); + await add("retry"); + let original: any; + await deliverAmplitude(db, config, async (_u, o) => { + original = JSON.parse(String(o?.body)).events[0]; + throw Error("network"); + }); + assert.equal((await states())[0].status, "PENDING"); + await db.query( + 'UPDATE business_events SET analytics_context=\'{"clientId":"111.222","sessionId":123}\'', + ); + await db.query( + "UPDATE amplitude_deliveries SET available_at=now()-interval '1 second'", + ); + await deliverAmplitude(db, config, async (u, o) => { + assert.deepEqual(JSON.parse(String(o?.body)).events[0], original); + return success(u, o); + }); + assert.equal((await states())[0].status, "INGESTED"); + assert.equal((await states())[0].attempts, 2); +}); +test("rate limits retry, invalid requests reject, stale attempts expire", async () => { + for (const [code, status] of [ + [429, "PENDING"], + [503, "PENDING"], + [400, "REJECTED"], + ] as const) { + await reset(); + await add("status-" + code); + await deliverAmplitude( + db, + config, + async () => new Response("{}", { status: code }), + ); + assert.equal((await states())[0].status, status); + } + await reset(); + await add("expired"); + await db.query( + "UPDATE amplitude_deliveries SET first_attempt_at=now()-interval '7 days'", + ); + await deliverAmplitude(db, config, async () => { + throw Error("must not send"); + }); + assert.equal((await states())[0].status, "EXPIRED"); +}); +test("expired lease can retry with stable payload; missing configuration does not send", async () => { + await reset(); + await add("lease"); + await db.query( + "UPDATE amplitude_deliveries SET status='PROCESSING',lease_until=now()-interval '1 second'", + ); + await deliverAmplitude(db, config, success); + assert.equal((await states())[0].status, "INGESTED"); + assert.equal(amplitudeConfig({}), undefined); + assert.throws(() => + amplitudeConfig({ + AMPLITUDE_API_KEY: "test-amplitude-key", + AMPLITUDE_REGION: "other", + }), + ); + assert.throws(() => amplitudeEvent({ type: "UNKNOWN" })); +}); diff --git a/apps/server/test/analytics.test.ts b/apps/server/test/analytics.test.ts new file mode 100644 index 0000000..1a9fd69 --- /dev/null +++ b/apps/server/test/analytics.test.ts @@ -0,0 +1,350 @@ +import { test, before, after } from "node:test"; +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { readFileSync } from "node:fs"; +import { database, type DB } from "../src/db.js"; +import { createApp } from "../src/app.js"; +import { seed } from "../src/content.js"; +import { businessEvent, visit } from "../src/business.js"; +import { + analyticsContext, + readAnalyticsContext, +} from "../src/analytics-context.js"; +import { ga4Payload, deliverGa4, ga4Config } from "../src/ga4.js"; +import { eventProperties, ga4EventName } from "../src/event-catalog.js"; +process.env.NODE_ENV = "test"; +let db: DB, + app: Awaited>, + actor: string, + cookie: string, + csrf: string, + analyticsId: string; +const origin = "http://localhost:5173"; +const config = { measurementId: "G-TEST12345", apiSecret: "test-only" }; +before(async () => { + db = await database(); + await seed(db); + app = await createApp({ + db, + origin, + googleClientId: "test", + sessionSecret: "test", + runnerToken: "test", + llmEnabled: false, + adminEmails: ["admin@example.test"], + verifyGoogle: async () => ({ + sub: "analytics-admin", + name: "Admin", + email: "admin@example.test", + }), + }); + const r = await app.inject({ + method: "POST", + url: "/api/auth/google", + headers: { origin }, + payload: { credential: "analytics-test-token" }, + }); + assert.equal(r.statusCode, 200, r.body); + cookie = String(r.headers["set-cookie"]).split(";")[0]; + csrf = r.json().csrf; + actor = r.json().user.id; + analyticsId = r.json().user.analyticsId; +}); +after(async () => { + await app.close(); + await db.close(); +}); +const reset = () => db.query("DELETE FROM business_events"); +const add = (key: string = randomUUID()) => + businessEvent(db, "LOGOUT", actor, null, key); +const statuses = async () => + ( + await db.query( + "SELECT status,last_code,attempts FROM analytics_deliveries ORDER BY event_id", + ) + ).rows; +const post = (events: any[], headers: Record = {}) => + app.inject({ + method: "POST", + url: "/api/analytics/events", + headers: { origin, cookie, "x-csrf-token": csrf, ...headers }, + payload: { events }, + }); +const okFetch: typeof fetch = async (url, options) => + new Response( + String(url).includes("/debug/") + ? JSON.stringify({ validationMessages: [] }) + : null, + { status: String(url).includes("/debug/") ? 200 : 204 }, + ); + +test("transactional outbox deduplicates, rolls back, and daily visits are one event per KST day", async () => { + await reset(); + await add("same"); + await add("same"); + assert.equal((await statuses()).length, 1); + await assert.rejects( + db.tx(async (tx) => { + await businessEvent(tx, "LOGOUT", actor, null, "rollback"); + throw Error("rollback"); + }), + ); + assert.equal((await statuses()).length, 1); + await db.query("DELETE FROM visitor_days"); + await visit(db, actor); + await visit(db, actor); + assert.equal( + ( + await db.query( + "SELECT count(*)::int AS n FROM business_events WHERE type='DAILY_VISIT'", + ) + ).rows[0].n, + 1, + ); +}); +test("client API rejects server-event spoofing, content fields, unknown topics and missing CSRF; duplicate retries do not double count", async () => { + await reset(); + const e = { + requestKey: randomUUID(), + type: "SCREEN_VIEWED", + properties: { screen: "home" }, + }; + assert.equal((await post([e], { "x-csrf-token": "" })).statusCode, 403); + assert.equal( + ( + await post([ + { ...e, type: "REVIEW_SUCCEEDED", properties: { model: "fake" } }, + ]) + ).statusCode, + 400, + ); + assert.equal( + ( + await post([ + { ...e, properties: { screen: "home", email: "private@example.test" } }, + ]) + ).statusCode, + 400, + ); + assert.equal( + ( + await post([ + e, + { + ...e, + requestKey: randomUUID(), + properties: { screen: "lesson", topic: "missing-topic" }, + }, + ]) + ).statusCode, + 400, + ); + assert.equal((await statuses()).length, 0); + const headers = { + "x-analytics-client-id": "123456.123456", + "x-analytics-session-id": "123456789", + }; + assert.equal((await post([e], headers)).statusCode, 200); + assert.equal((await post([e], headers)).statusCode, 200); + assert.equal((await statuses()).length, 1); + const row = (await db.query("SELECT * FROM business_events")).rows[0]; + assert.deepEqual(row.analytics_context, { + clientId: "123456.123456", + sessionId: 123456789, + }); + assert.equal((await app.inject("/api/admin/analytics")).statusCode, 401); + const status = await app.inject({ + url: "/api/admin/analytics", + headers: { cookie }, + }); + assert.equal(status.statusCode, 200); + assert.equal( + status.json().catalog.length, + Object.keys(eventProperties).length, + ); +}); +test("GA4 payload excludes raw identity, entity IDs, extra properties and matches browser context", () => { + const payload = ga4Payload({ + id: randomUUID(), + type: "SCREEN_VIEWED", + created_at: new Date(), + analytics_id: analyticsId, + actor_id: actor, + entity_id: "private-record", + email: "private@example.test", + metadata: { screen: "home", code: "private code" }, + analytics_context: { clientId: "123.456", sessionId: 123 }, + }); + assert.equal(payload.client_id, "123.456"); + assert.equal(payload.user_id, analyticsId); + assert.equal(payload.events[0].params.session_id, 123); + assert.doesNotMatch(JSON.stringify(payload), /private|actor_id|entity_id/); + assert.deepEqual( + readAnalyticsContext({ + "x-analytics-client-id": "someone@example.test", + "x-analytics-session-id": "NaN", + }), + {}, + ); + for (const type of Object.keys(eventProperties)) + assert.ok(/^[a-z][a-z0-9_]{0,39}$/.test(ga4EventName(type))); + assert.equal(ga4Config({}), undefined); + assert.throws(() => + ga4Config({ GA4_MEASUREMENT_ID: "bad", GA4_API_SECRET: "secret" }), + ); +}); +test("GA4 server-only client ID is numeric, stable and does not invent session attribution", () => { + const row = { + id: randomUUID(), + type: "LOGOUT", + created_at: new Date(), + analytics_id: analyticsId, + metadata: {}, + }; + const payload = ga4Payload(row); + assert.match(payload.client_id, /^\d+\.\d+$/); + assert.equal( + ga4Payload({ ...row, id: randomUUID() }).client_id, + payload.client_id, + ); + assert.notEqual( + ga4Payload({ ...row, analytics_id: randomUUID() }).client_id, + payload.client_id, + ); + assert.equal(payload.events[0].params.event_source, "server_unattributed"); + assert.equal(payload.events[0].params.session_id, undefined); + assert.equal(payload.events[0].params.engagement_time_msec, undefined); +}); + +test("async server result inherits the originating request context", async () => { + await reset(); + const entity = randomUUID(); + await analyticsContext.run({ clientId: "123.456", sessionId: 123 }, () => + businessEvent(db, "EXECUTION_REQUESTED", actor, entity, "req", { + mode: "judge", + language: "python", + problem: "sample", + }), + ); + await businessEvent(db, "EXECUTION_FINISHED", actor, entity, "done", { + status: "SUCCEEDED", + verdict: "ACCEPTED", + }); + const row = ( + await db.query( + "SELECT analytics_context FROM business_events WHERE event_key='done'", + ) + ).rows[0]; + assert.deepEqual(row.analytics_context, { + clientId: "123.456", + sessionId: 123, + }); +}); +test("GA4 batches at most 25, preserves original timestamps and validates before collecting", async () => { + await reset(); + for (let i = 0; i < 27; i++) await add(); + const calls: { url: string; body: any }[] = []; + const send: typeof fetch = async (url, options) => { + calls.push({ url: String(url), body: JSON.parse(options!.body as string) }); + return okFetch(url, options); + }; + assert.equal((await deliverGa4(db, config, send)).processed, 27); + assert.equal(calls.length, 4); + assert.match(calls[0].url, /\/debug\//); + assert.doesNotMatch(calls[1].url, /\/debug\//); + assert.equal(calls[0].body.events.length, 25); + assert.equal(calls[2].body.events.length, 2); + assert.ok( + calls[0].body.events.every((e: any) => + Number.isSafeInteger(e.timestamp_micros), + ), + ); + assert.ok((await statuses()).every((r) => r.status === "ACCEPTED")); + assert.equal((await deliverGa4(db, config, send)).processed, 0); +}); +test("validation rejects never collect and ambiguous delivery never retries", async () => { + await reset(); + await add(); + let calls = 0; + await deliverGa4(db, config, async () => { + calls++; + return Response.json({ + validationMessages: [{ validationCode: "VALUE_INVALID" }], + }); + }); + assert.equal(calls, 1); + assert.equal((await statuses())[0].status, "REJECTED"); + await reset(); + await add(); + calls = 0; + await deliverGa4(db, config, async (url, options) => { + calls++; + if (String(url).includes("/debug/")) return okFetch(url, options); + throw Error("simulated response loss"); + }); + assert.equal((await statuses())[0].status, "UNCERTAIN"); + assert.equal(calls, 2); + await deliverGa4(db, config, async () => { + throw Error("must not retry"); + }); + assert.equal((await statuses())[0].attempts, 1); +}); +test("pre-collect transport failure backs off; disabled, expired and abandoned leases never collect", async () => { + await reset(); + await add(); + await deliverGa4(db, undefined, async () => { + throw Error("disabled"); + }); + assert.equal((await statuses())[0].attempts, 0); + await deliverGa4(db, config, async () => { + throw Error("validation unavailable"); + }); + assert.equal((await statuses())[0].status, "PENDING"); + assert.equal((await statuses())[0].last_code, "VALIDATION_UNAVAILABLE"); + assert.equal((await deliverGa4(db, config, okFetch)).processed, 0); + await db.query( + "UPDATE business_events SET created_at=now()-interval '73 hours'", + ); + await deliverGa4(db, config, okFetch); + assert.equal((await statuses())[0].status, "EXPIRED"); + await reset(); + await add(); + await db.query( + "UPDATE analytics_deliveries SET status='PROCESSING',lease_until=now()-interval '1 second'", + ); + await deliverGa4(db, config, async () => { + throw Error("must not resend"); + }); + assert.equal((await statuses())[0].status, "UNCERTAIN"); +}); +test("Internal analytics views expose pseudonymous data, all dashboard SQL runs, PUBLIC cannot select views", async () => { + await reset(); + await add(); + const row = (await db.query("SELECT * FROM analytics_events")).rows[0]; + assert.equal(row.user_id, analyticsId); + assert.notEqual(row.user_id, actor); + assert.doesNotMatch( + JSON.stringify(row), + /admin@example|actor_id|analytics_context|entity_id/, + ); + for (const q of JSON.parse( + readFileSync( + new URL("../../../scripts/business-queries.json", import.meta.url), + "utf8", + ), + )) { + // Production uses algostep; the isolated database uses public. + await db.query( + q.query.replaceAll("algostep.analytics_", "public.analytics_"), + ); + } + await db.query("CREATE ROLE analytics_test_reader"); + assert.equal( + ( + await db.query( + "SELECT has_table_privilege('analytics_test_reader','analytics_events','SELECT') AS allowed", + ) + ).rows[0].allowed, + false, + ); +}); diff --git a/apps/server/test/expansion.test.ts b/apps/server/test/expansion.test.ts index 0aeae42..e2c5387 100644 --- a/apps/server/test/expansion.test.ts +++ b/apps/server/test/expansion.test.ts @@ -28,7 +28,7 @@ test("migration is repeatable and preserves saved data", async () => { assert.equal( (await db.query("SELECT max(version) AS n FROM schema_migrations")) .rows[0].n, - 5, + 7, ); } finally { await db.close(); diff --git a/apps/web/src/AdminDashboard.tsx b/apps/web/src/AdminDashboard.tsx index ffce7c2..86fb23f 100644 --- a/apps/web/src/AdminDashboard.tsx +++ b/apps/web/src/AdminDashboard.tsx @@ -1,3 +1,4 @@ +import { track } from "./analytics.js"; import React, { useEffect, useState } from "react"; import { api, post } from "./api.js"; import { SupportPanel, Pager, displayTime } from "./SupportPanel.js"; @@ -33,6 +34,9 @@ export function AdminDashboard() { [data, setData] = useState(null), [error, setError] = useState(""), [refresh, setRefresh] = useState(0); + useEffect(() => { + track("ADMIN_SECTION_VIEWED", { section: tab }); + }, [tab]); useEffect(() => { let current = true; setData(null); @@ -438,6 +442,7 @@ function EventList({ query, events }: { query: string; events: any[] }) { const [type, setType] = useState(""); return ( <> +