root/sp
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(phase6a-v1.2): dual-mode custom-event enrichment (v1.0 + v2.0 aliases)

Browse Source 
- Prisma: posthog_person_cache (24h TTL), posthog_group_cache, session_custom_events
- SessionMeta: +posthogDistinctId, +groupKey, +customEventCount
- PostHog client: listSessionEvents, getPerson, getGroup
- posthog-cache.ts: cache-with-stale-refresh for person + group properties
- event-taxonomy.ts: dual-mode alias map (vin_decoded↔vin_decode_initiated,
  vin_decode_success↔vin_decode_succeeded, vin_decode_error↔vin_decode_failed,
  payment_success↔payment_succeeded). 56 tracked event names total.
- enrich.ts: per-session custom events fetch, persist to session_custom_events,
  attach user_properties + group_properties via cache.
- tagger.ts: 14 new custom-event tags
  (vin_decode_fail_pattern, provider_reliability_issue, payment_friction,
   payment_ui_silent_failure (P0), payment_failed_session, checkout_abandonment,
   downgrade_pending, compatibility_quality_gap, parts_export_abandoned,
   search_validation_friction, onboarding_stuck, upgrade_hesitation,
   power_user_path, at_risk_active_session, provider_mismatch,
   api_key_friction, webhook_setup_struggle); group-context priority bump.
- heuristic.ts: customEventPromoteReasons() — 10 custom-event promote signals.
- compress.ts: merge_rrweb_and_custom_events interleaving with ⤷ EVENT: prefix,
  ±1.5s grouping, standalone flush for events with no nearby rrweb. Adds
  company_context header line; key_events lists top-5 custom event counts.
- /insights page: +Events column. /insights/sessions/[id]: collapsible custom
  events table + group key display.

Graceful: when Sase.tr emits 0 custom events the pipeline falls back to
rrweb-only tagging from Phase 6a v1.1.

PRD: behavioral-insight-pipeline-prd-1.md (v1.2),
     sase-posthog-events-prd.md (Sase.tr-side, separate codebase).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Semih
2026-05-13 22:17:18 +00:00
parent 93499faeaa
commit fc6a3f7a39
16 changed files with 5059 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
apps/worker/src/jobs/compress-sessions.ts
View File

@@ -2,6 +2,8 @@ import { prisma } from "../db";
import { getSnapshotSources, getSnapshotBlob, isConfigured } from "../lib/posthog";
import { compressSnapshots, parseSnapshotBlob } from "../lib/compress";
import { putText } from "../lib/minio";
import { loadEnrichment } from "../lib/enrich";
import { getCachedGroup } from "../lib/posthog-cache";
const COMPRESSION_BUCKET = process.env.INSIGHT_COMPRESSED_BUCKET ?? "insight-compressed";
@@ -35,18 +37,38 @@ export async function runCompressSessions(): Promise<{ compressed: number; faile
const blob = await getSnapshotBlob(s.id, source, sorted[0], sorted[sorted.length - 1]);
events.push(...parseSnapshotBlob(blob));
}
const out = compressSnapshots(events, {
sessionId: s.id,
projectKey: s.projectKey,
userSegment: `${s.isAuthenticated ? "auth" : "anon"}${
s.subscriptionTier ? `, tier=${s.subscriptionTier}` : ""
}`,
duration: humanDuration(s.durationMs),
tags: s.tags,
severity: s.severity ?? "INFO",
score: s.score ?? 0,
startUrl: s.startUrl,
});
const { customEvents } = await loadEnrichment(s.id);
let companyContext: string | null = null;
if (s.groupKey) {
const group = await getCachedGroup("company", s.groupKey).catch(() => null);
if (group) {
const p = group.properties;
const parts: string[] = [];
if (p.tier) parts.push(`tier=${p.tier}`);
if (typeof p.mrr_usd === "number") parts.push(`mrr=$${p.mrr_usd}`);
if (typeof p.seat_count === "number") parts.push(`seats=${p.seats_used ?? "?"}/${p.seat_count}`);
if (p.industry) parts.push(`industry=${p.industry}`);
companyContext = parts.join(", ") || null;
}
}
const out = compressSnapshots(
events,
{
sessionId: s.id,
projectKey: s.projectKey,
userSegment: `${s.isAuthenticated ? "auth" : "anon"}${
s.subscriptionTier ? `, tier=${s.subscriptionTier}` : ""
}`,
duration: humanDuration(s.durationMs),
tags: s.tags,
severity: s.severity ?? "INFO",
score: s.score ?? 0,
startUrl: s.startUrl,
companyContext,
},
customEvents,
);
// Sanitization sanity check: large token output but zero matches is an anomaly.
// Don't abort here (Phase 6a defers LLM), but flag it.

1
apps/worker/src/jobs/posthog-ingest.ts
View File

@@ -92,6 +92,7 @@ export async function runPostHogIngest(): Promise<IngestResult> {
id: rec.id,
projectKey: PROJECT_KEY,
userIdHash: hashUserId(rec.distinct_id),
posthogDistinctId: rec.distinct_id,
isAuthenticated: isAuth,
subscriptionTier: tier,
startedAt,

43
apps/worker/src/jobs/tag-sessions.ts
View File

@@ -1,13 +1,16 @@
import { prisma } from "../db";
import { tagSession, scoreSession } from "../lib/tagger";
import { buildEnrichment } from "../lib/enrich";
import { customEventPromoteReasons } from "../lib/heuristic";
const MIN_SCORE_FOR_COMPRESSION = Number(process.env.INSIGHT_MIN_SCORE ?? "30");
const TAG_BATCH_SIZE = Number(process.env.INSIGHT_TAG_BATCH ?? "40");
export async function runTagSessions(): Promise<{ tagged: number; discarded: number }> {
const pending = await prisma.sessionMeta.findMany({
where: { status: "pending_signal" },
orderBy: { startedAt: "asc" },
take: 200,
take: TAG_BATCH_SIZE,
});
if (pending.length === 0) return { tagged: 0, discarded: 0 };
@@ -15,17 +18,40 @@ export async function runTagSessions(): Promise<{ tagged: number; discarded: num
let discarded = 0;
for (const s of pending) {
// Sase RO enrichment (onboarding_stuck) deferred: distinct_id is hashed in panel,
// future enrichment can flow via PostHog person properties → session_meta extension.
const result = tagSession(s, null);
// Enrichment is best-effort — degrade gracefully if PostHog events endpoint is unavailable.
const endedAt = new Date(s.startedAt.getTime() + s.durationMs);
let enrichment;
try {
enrichment = await buildEnrichment({
sessionId: s.id,
distinctId: s.posthogDistinctId,
startedAt: s.startedAt,
endedAt,
});
} catch (e) {
console.warn(`[tag] enrichment failed ${s.id}: ${(e as Error).message}`);
enrichment = { customEvents: [], userProperties: {}, groupProperties: null, groupKey: null };
}
// Novelty/pattern still rough — pre-LLM phase
const score = scoreSession(s, result.severity, 0.5, 0.2);
// Add custom-event-derived promote reasons to the existing rrweb reasons.
const eventNames = enrichment.customEvents.map((e) => e.name);
const extraReasons = customEventPromoteReasons(eventNames);
const mergedReasons = Array.from(new Set([...s.promotionReasons, ...extraReasons]));
const result = tagSession(s, enrichment);
const score = scoreSession(s, result.severity, 0.5, 0.2, enrichment.groupProperties);
if (result.tags.length === 0 || score < MIN_SCORE_FOR_COMPRESSION) {
await prisma.sessionMeta.update({
where: { id: s.id },
data: { status: "discarded", processedAt: new Date(), score },
data: {
status: "discarded",
processedAt: new Date(),
score,
customEventCount: enrichment.customEvents.length,
promotionReasons: mergedReasons,
groupKey: enrichment.groupKey,
},
});
discarded++;
continue;
@@ -39,6 +65,9 @@ export async function runTagSessions(): Promise<{ tagged: number; discarded: num
score,
status: "tagged",
processedAt: new Date(),
customEventCount: enrichment.customEvents.length,
promotionReasons: mergedReasons,
groupKey: enrichment.groupKey,
},
});
tagged++;

87
apps/worker/src/lib/compress.ts
View File

@@ -3,6 +3,7 @@
import { sanitize, type SanitizationReport } from "./sanitize";
import { fingerprintHash } from "./hash";
import type { CanonicalEvent } from "./event-taxonomy";
export type CompressedOutput = {
timeline: string;
@@ -28,6 +29,7 @@ type SessionHeader = {
severity: string;
score: number;
startUrl: string | null;
companyContext?: string | null;
};
const MAX_LINES = 80;
@@ -35,10 +37,17 @@ const MAX_LINES = 80;
export function compressSnapshots(
events: RREvent[],
header: SessionHeader,
customEvents: CanonicalEvent[] = [],
): CompressedOutput {
// Sort by timestamp
events.sort((a, b) => a.timestamp - b.timestamp);
const t0 = events[0]?.timestamp ?? 0;
const t0 = events[0]?.timestamp ?? customEvents[0]?.timestamp.getTime() ?? 0;
// Convert custom events to ms-based timestamps and bucket by second for interleaving
const customTs: Array<{ ts: number; ev: CanonicalEvent }> = customEvents
.map((ev) => ({ ts: ev.timestamp.getTime(), ev }))
.sort((a, b) => a.ts - b.ts);
let customIdx = 0;
const lines: string[] = [];
let lastClickAt = 0;
@@ -51,11 +60,63 @@ export function compressSnapshots(
const failedEndpoints: string[] = [];
let url: string | null = header.startUrl;
const formatCustom = (ev: CanonicalEvent): string => {
// Inline a tiny subset of important properties to keep tokens bounded.
const p = ev.properties ?? {};
const keys = [
"provider",
"provider_attempted",
"error_code",
"result",
"plan",
"amount",
"currency",
"vin_brand",
"query_source",
"filter_type",
"format",
"feature_name",
"reason",
"method",
];
const parts: string[] = [];
for (const k of keys) {
const v = p[k];
if (v === undefined || v === null) continue;
const sv = typeof v === "string" ? `"${truncate(v, 30)}"` : String(v);
parts.push(`${k}:${sv}`);
}
return ` ⤷ EVENT: ${ev.name}${parts.length ? ` { ${parts.join(", ")} }` : ""}`;
};
const flushCustomUpTo = (rrwebMs: number, attach: boolean) => {
while (customIdx < customTs.length) {
const c = customTs[customIdx];
// Attach if within 1.5s of the last rrweb line, otherwise stop and let the next rrweb line handle.
if (attach && Math.abs(c.ts - rrwebMs) <= 1500) {
if (lines.length < MAX_LINES) lines.push(formatCustom(c.ev));
customIdx++;
continue;
}
if (!attach && c.ts <= rrwebMs) {
// Emit standalone at the relative time of the custom event
const tRel = Math.max(0, Math.round((c.ts - t0) / 1000));
if (lines.length < MAX_LINES)
lines.push(`${formatTs(tRel)} → ${formatCustom(c.ev).trim().replace(/^⤷ /, "")}`);
customIdx++;
continue;
}
break;
}
};
for (const ev of events) {
if (lines.length >= MAX_LINES) {
lines.push(`... (timeline truncated at ${MAX_LINES} entries)`);
break;
}
// Emit any standalone custom events that occurred before this rrweb event (>1.5s gap)
flushCustomUpTo(ev.timestamp - 1500, false);
const tRel = Math.max(0, Math.round((ev.timestamp - t0) / 1000));
const ts = formatTs(tRel);
@@ -126,7 +187,8 @@ export function compressSnapshots(
// Custom (PostHog autocapture events)
const tag = ev.data?.tag;
if (tag === "$autocapture") {
continue; // already covered by rrweb type 3
flushCustomUpTo(ev.timestamp, true);
continue;
}
const payload = ev.data?.payload;
if (tag === "$exception" || tag === "console_error") {
@@ -134,8 +196,18 @@ export function compressSnapshots(
errors.push(msg);
lines.push(`${ts} → ⚠ console_error "${msg}"`);
}
flushCustomUpTo(ev.timestamp, true);
continue;
}
// After each handled rrweb event, attach any custom events within ±1.5s
flushCustomUpTo(ev.timestamp, true);
}
// Flush any remaining custom events as standalone at the end
while (customIdx < customTs.length && lines.length < MAX_LINES) {
const c = customTs[customIdx++];
const tRel = Math.max(0, Math.round((c.ts - t0) / 1000));
lines.push(`${formatTs(tRel)} → ${formatCustom(c.ev).trim().replace(/^⤷ /, "")}`);
}
// Sanitize the assembled timeline as a final defense.
@@ -146,6 +218,16 @@ export function compressSnapshots(
if (network5xx.length) keyEvents.push(`${network5xx.length}x 5xx on ${[...new Set(failedEndpoints)].slice(0, 3).join(", ")}`);
if (errors.length) keyEvents.push(`${errors.length}x console error`);
if (rageEmitted) keyEvents.push("rage click detected");
if (customEvents.length) {
const counts: Record<string, number> = {};
for (const c of customEvents) counts[c.name] = (counts[c.name] ?? 0) + 1;
const top = Object.entries(counts)
.sort((a, b) => b[1] - a[1])
.slice(0, 5)
.map(([n, c]) => `${c}x ${n}`)
.join(", ");
keyEvents.push(`custom events: ${top}`);
}
if (!keyEvents.length) keyEvents.push("no notable signals");
const hypotheses: string[] = [];
@@ -161,6 +243,7 @@ export function compressSnapshots(
`id: ${header.sessionId}`,
`project: ${header.projectKey}`,
`user_segment: ${header.userSegment}`,
...(header.companyContext ? [`company_context: ${header.companyContext}`] : []),
`duration: ${header.duration}`,
`tags: [${header.tags.join(", ")}]`,
`severity: ${header.severity}`,

96
apps/worker/src/lib/enrich.ts Normal file
View File

@@ -0,0 +1,96 @@
import { prisma } from "./../db";
import { listSessionEvents } from "./posthog";
import { getCachedPerson, getCachedGroup } from "./posthog-cache";
import { normalizeEvents, TRACKED_EVENTS, type CanonicalEvent } from "./event-taxonomy";
export type EnrichedContext = {
customEvents: CanonicalEvent[];
userProperties: Record<string, unknown>;
groupProperties: Record<string, unknown> | null;
groupKey: string | null;
};
const COMPANY_GROUP_TYPE = process.env.POSTHOG_COMPANY_GROUP_TYPE ?? "company";
// Build enrichment for a single session. Distinct_id should be the raw PostHog distinct_id
// (we keep that on the SessionMeta-side in a separate pass — see ingest).
export async function buildEnrichment(opts: {
sessionId: string;
distinctId: string | null;
startedAt: Date;
endedAt: Date;
}): Promise<EnrichedContext> {
const out: EnrichedContext = {
customEvents: [],
userProperties: {},
groupProperties: null,
groupKey: null,
};
// Fetch custom events for this session
try {
// Widen the window slightly to catch late-flushed events
const dateFrom = new Date(opts.startedAt.getTime() - 60_000).toISOString();
const dateTo = new Date(opts.endedAt.getTime() + 5 * 60_000).toISOString();
const raw = await listSessionEvents({
sessionId: opts.sessionId,
dateFrom,
dateTo,
eventNames: TRACKED_EVENTS,
});
out.customEvents = normalizeEvents(raw);
} catch (e) {
console.warn(`[enrich] events fetch failed for ${opts.sessionId}: ${(e as Error).message}`);
}
// Persist custom events for compression timeline interleaving
if (out.customEvents.length) {
// Clear + replace (idempotent for re-runs)
await prisma.sessionCustomEvent.deleteMany({ where: { sessionId: opts.sessionId } });
await prisma.sessionCustomEvent.createMany({
data: out.customEvents.map((e) => ({
sessionId: opts.sessionId,
eventName: e.name,
timestamp: e.timestamp,
properties: e.properties as object,
})),
});
}
// Person properties cache
if (opts.distinctId) {
const person = await getCachedPerson(opts.distinctId).catch(() => null);
if (person) {
out.userProperties = person.properties;
if (person.groups && person.groups[COMPANY_GROUP_TYPE]) {
out.groupKey = person.groups[COMPANY_GROUP_TYPE];
}
}
}
// Group properties (company)
if (out.groupKey) {
const group = await getCachedGroup(COMPANY_GROUP_TYPE, out.groupKey).catch(() => null);
if (group) out.groupProperties = group.properties;
}
return out;
}
// Load the persisted enrichment for a session (used by compression).
export async function loadEnrichment(sessionId: string): Promise<{
customEvents: CanonicalEvent[];
}> {
const rows = await prisma.sessionCustomEvent.findMany({
where: { sessionId },
orderBy: { timestamp: "asc" },
});
return {
customEvents: rows.map((r) => ({
name: r.eventName,
rawName: r.eventName,
timestamp: r.timestamp,
properties: (r.properties as Record<string, unknown>) ?? {},
})),
};
}

123
apps/worker/src/lib/event-taxonomy.ts Normal file
View File

@@ -0,0 +1,123 @@
// Dual-mode event taxonomy: Sase.tr is migrating from v1.0 to v2.0 names.
// Both must work in parallel. We resolve to a canonical name for tagging logic.
const ALIASES: Record<string, string> = {
// VIN decode family
vin_decoded: "vin_decode_initiated",
vin_decode_success: "vin_decode_succeeded",
vin_decode_error: "vin_decode_failed",
// Payment dual-capture rename
payment_success: "payment_succeeded", // backend canonical
// Note: payment_success_ui (frontend) is a new name with no v1 alias; left as-is.
// Other v1 → v2 expected renames; safe identity if not yet emitted
// (kept for forward compat — if Sase.tr stays on v1, identity also works)
};
export function canonical(name: string): string {
return ALIASES[name] ?? name;
}
// All event names we want to fetch from PostHog (both v1 + v2 variants).
export const TRACKED_EVENTS: string[] = [
// VIN decode (v1 + v2)
"vin_decoded",
"vin_decode_initiated",
"vin_decode_success",
"vin_decode_succeeded",
"vin_decode_error",
"vin_decode_failed",
"vin_decode_candidates",
"vin_decode_candidates_shown",
"vin_decode_candidate_selected",
// Provider
"provider_response_received",
"provider_fallback_triggered",
"provider_data_quality_flag",
"multi_provider_search_initiated",
// Parts / compatibility
"parts_panel_viewed",
"parts_filter_applied",
"compatibility_check_initiated",
"compatibility_check_completed",
"parts_export_initiated",
"parts_export_completed",
"oem_code_copied",
// Payment (v1 + v2)
"payment_initiated",
"payment_success",
"payment_succeeded",
"payment_success_ui",
"payment_failed",
"payment_failed_ui",
"receipt_uploaded",
// Subscription funnel
"plan_selected",
"yearly_toggle_clicked",
"checkout_started",
"trial_started",
"subscription_cancelled",
"subscription_resumed",
"cancel_flow_viewed",
"cancel_save_clicked",
"cancel_save_offer_accepted",
"downgrade_offer_shown",
"downgrade_offer_accepted",
"downgrade_offer_declined",
// Trial urgency banner
"trial_urgency_banner_viewed",
"trial_urgency_banner_cta_clicked",
"trial_urgency_banner_dismissed",
// Search behavior
"search_input_focused",
"search_input_validation_failed",
"search_paste_detected",
"search_history_opened",
"search_history_item_selected",
// Feature discovery & help
"feature_discovered",
"help_clicked",
"docs_link_clicked",
"tutorial_started",
"tutorial_step_completed",
"tutorial_abandoned",
// API key / webhook
"api_key_created",
"api_key_revoked",
"webhook_configured",
"webhook_delivery_failed",
// Auth lifecycle
"user_signed_up",
"user_logged_in",
"user_logged_out",
];
export type CanonicalEvent = {
name: string; // canonical (v2-style)
rawName: string; // as emitted
timestamp: Date;
properties: Record<string, unknown>;
};
export function normalizeEvents(
events: Array<{ event: string; timestamp: string; properties: Record<string, unknown> }>,
): CanonicalEvent[] {
return events
.map((e) => ({
name: canonical(e.event),
rawName: e.event,
timestamp: new Date(e.timestamp),
properties: e.properties ?? {},
}))
.sort((a, b) => a.timestamp.getTime() - b.timestamp.getTime());
}

38
apps/worker/src/lib/heuristic.ts
View File

@@ -66,6 +66,42 @@ export function inferAuthenticated(rec: PHRecordingListItem): boolean {
export function inferSubscriptionTier(rec: PHRecordingListItem): string | null {
const props = rec.person?.properties ?? {};
const v = props.subscription_tier ?? props.tier ?? props.plan;
const v = props.subscription_tier ?? props.tier ?? props.plan_tier ?? props.plan;
return typeof v === "string" ? v : null;
}
// Promote signals derived from custom events (PRD v1.2 Bölüm 5.5).
// Called after initial heuristic pass, can flip a session from discard to keep.
export function customEventPromoteReasons(eventNames: string[]): string[] {
const set = new Set(eventNames);
const reasons: string[] = [];
// High-value semantic signals (single occurrence is enough)
const single = [
"vin_decode_failed",
"vin_decode_error",
"provider_fallback_triggered",
"payment_failed",
"subscription_cancelled",
"trial_urgency_banner_cta_clicked",
"downgrade_offer_shown",
];
for (const e of single) if (set.has(e)) reasons.push(`event:${e}`);
// Funnel-state signals
if (set.has("checkout_started") && !set.has("payment_succeeded") && !set.has("payment_success"))
reasons.push("event:checkout_no_pay");
if (
(set.has("trial_urgency_banner_cta_clicked") || set.has("trial_started")) &&
!set.has("payment_succeeded") &&
!set.has("payment_success")
)
reasons.push("event:trial_no_pay");
// Compatibility quality
// (the specific 'result' check is not visible here — only event presence; full check in tagger)
if (set.has("compatibility_check_completed")) reasons.push("event:compat_check");
return reasons;
}

96
apps/worker/src/lib/posthog-cache.ts Normal file
View File

@@ -0,0 +1,96 @@
import { prisma } from "../db";
import { getPerson, getGroup } from "./posthog";
const TTL_HOURS = Number(process.env.POSTHOG_CACHE_TTL_HOURS ?? "24");
export type PersonCache = {
properties: Record<string, unknown>;
groups: Record<string, string> | null;
};
export type GroupCache = {
properties: Record<string, unknown>;
};
export async function getCachedPerson(distinctId: string): Promise<PersonCache | null> {
const now = new Date();
const row = await prisma.posthogPersonCache.findUnique({ where: { distinctId } });
if (row && row.ttlAt > now) {
return {
properties: (row.properties as Record<string, unknown>) ?? {},
groups: (row.groups as Record<string, string> | null) ?? null,
};
}
// Stale or missing → refresh.
let person;
try {
person = await getPerson(distinctId);
} catch {
return row
? {
properties: (row.properties as Record<string, unknown>) ?? {},
groups: (row.groups as Record<string, string> | null) ?? null,
}
: null;
}
if (!person) return null;
const ttlAt = new Date(now.getTime() + TTL_HOURS * 3600_000);
const groups: Record<string, string> = {};
// PostHog person properties may include $groups field (depends on plan)
const phGroups = (person.properties as any)?.$groups;
if (phGroups && typeof phGroups === "object") {
for (const [k, v] of Object.entries(phGroups)) {
if (typeof v === "string") groups[k] = v;
}
}
const data = {
properties: person.properties ?? {},
groups: Object.keys(groups).length ? groups : null,
};
const groupsJson = (data.groups ?? null) as object | null;
await prisma.posthogPersonCache.upsert({
where: { distinctId },
create: {
distinctId,
properties: data.properties as object,
...(groupsJson ? { groups: groupsJson } : {}),
refreshedAt: now,
ttlAt,
},
update: {
properties: data.properties as object,
...(groupsJson ? { groups: groupsJson } : { groups: undefined }),
refreshedAt: now,
ttlAt,
},
});
return data;
}
export async function getCachedGroup(
groupType: string,
groupKey: string,
): Promise<GroupCache | null> {
const now = new Date();
const row = await prisma.posthogGroupCache.findUnique({
where: { groupType_groupKey: { groupType, groupKey } },
});
if (row && row.ttlAt > now) {
return { properties: (row.properties as Record<string, unknown>) ?? {} };
}
let group;
try {
group = await getGroup(groupType, groupKey);
} catch {
return row ? { properties: (row.properties as Record<string, unknown>) ?? {} } : null;
}
if (!group) return null;
const ttlAt = new Date(now.getTime() + TTL_HOURS * 3600_000);
const data = { properties: group.group_properties };
await prisma.posthogGroupCache.upsert({
where: { groupType_groupKey: { groupType, groupKey } },
create: { groupType, groupKey, properties: data.properties as object, refreshedAt: now, ttlAt },
update: { properties: data.properties as object, refreshedAt: now, ttlAt },
});
return data;
}

78
apps/worker/src/lib/posthog.ts
View File

@@ -98,3 +98,81 @@ export async function getRecording(sessionId: string): Promise<PHRecordingListIt
export function isConfigured(): boolean {
return Boolean(TOKEN && PROJECT_ID);
}
// ---------- Custom events ----------
export type PHCustomEvent = {
event: string;
timestamp: string;
distinct_id: string;
properties: Record<string, unknown>;
};
// Returns events for a single session, sorted ascending by timestamp.
// `eventNames` filters server-side via PostHog event= param when supported.
export async function listSessionEvents(opts: {
sessionId: string;
dateFrom: string;
dateTo: string;
eventNames?: string[];
}): Promise<PHCustomEvent[]> {
const url = new URL(`${HOST}/api/projects/${PROJECT_ID}/events/`);
url.searchParams.set("limit", "200");
url.searchParams.set("after", opts.dateFrom);
url.searchParams.set("before", opts.dateTo);
url.searchParams.set(
"properties",
JSON.stringify([{ key: "$session_id", value: opts.sessionId, operator: "exact" }]),
);
if (opts.eventNames && opts.eventNames.length) {
// PostHog accepts repeated event= or comma-joined event=
for (const n of opts.eventNames) url.searchParams.append("event", n);
}
const res = await fetch(url, { headers: headers() });
if (!res.ok) throw new Error(`posthog events ${res.status}`);
const data = (await res.json()) as { results?: PHCustomEvent[] };
const out = data.results ?? [];
out.sort((a, b) => new Date(a.timestamp).getTime() - new Date(b.timestamp).getTime());
return out;
}
// ---------- Person & group fetch ----------
export type PHPerson = {
distinct_ids: string[];
properties: Record<string, unknown>;
// Person group memberships, when available
// (PostHog only exposes via separate /persons/{id}/groups endpoint depending on plan)
};
export async function getPerson(distinctId: string): Promise<PHPerson | null> {
const url = new URL(`${HOST}/api/projects/${PROJECT_ID}/persons/`);
url.searchParams.set("distinct_id", distinctId);
const res = await fetch(url, { headers: headers() });
if (res.status === 404) return null;
if (!res.ok) throw new Error(`posthog person ${res.status}`);
const data = (await res.json()) as { results?: PHPerson[] };
return data.results?.[0] ?? null;
}
export async function getGroup(
groupType: string,
groupKey: string,
): Promise<{ group_properties: Record<string, unknown> } | null> {
const url = new URL(`${HOST}/api/projects/${PROJECT_ID}/groups/`);
url.searchParams.set("group_type_index", "0"); // PostHog uses indexes 0..4; resolved below if needed
url.searchParams.set("group_key", groupKey);
const res = await fetch(url, { headers: headers() });
if (res.status === 404) return null;
if (!res.ok) {
// Group types are indexed; fall back to no-op gracefully so callers can degrade.
return null;
}
const data = (await res.json()) as { results?: Array<{ group_properties?: Record<string, unknown> }> };
const row = data.results?.[0];
if (!row) return null;
return { group_properties: row.group_properties ?? {} };
// Note: groupType is currently informational; group_type_index resolution can be added when
// the project has >1 group type. For Sase.tr only 'company' is expected.
void groupType;
}

213
apps/worker/src/lib/tagger.ts
View File

@@ -1,4 +1,5 @@
import type { SessionMeta } from "@prisma/client";
import type { CanonicalEvent } from "./event-taxonomy";
export type SeverityLevel = "P0" | "P1" | "P2" | "P3" | "INFO";
@@ -19,60 +20,217 @@ export type TagResult = {
severity: SeverityLevel;
};
export type OnboardingContext = {
signupRecent: boolean; // signup_at < 24h
firstVinQuery: boolean;
export type EnrichmentCtx = {
customEvents: CanonicalEvent[];
userProperties: Record<string, unknown>;
groupProperties: Record<string, unknown> | null;
};
export function tagSession(s: SessionMeta, ctx: OnboardingContext | null): TagResult {
function has(events: CanonicalEvent[], name: string): boolean {
return events.some((e) => e.name === name);
}
function count(events: CanonicalEvent[], name: string): number {
return events.filter((e) => e.name === name).length;
}
function pick<T extends string | number | boolean>(
obj: Record<string, unknown>,
key: string,
): T | undefined {
const v = obj[key];
return v as T | undefined;
}
export function tagSession(s: SessionMeta, ctx: EnrichmentCtx | null): TagResult {
const tags: string[] = [];
let severity: SeverityLevel = "INFO";
const events = ctx?.customEvents ?? [];
const userProps = ctx?.userProperties ?? {};
const groupProps = ctx?.groupProperties ?? null;
// bug_suspected
// ─── Bug detection (rrweb-based, generic fallback) ───
if (s.errorCount > 0 && (s.rageClickCount > 0 || s.network5xxCount > 0)) {
tags.push("bug_suspected");
severity = bump(severity, "P1");
}
// server_error_impact
if (s.network5xxCount > 2) {
tags.push("server_error_impact");
severity = bump(severity, "P1");
}
// ux_friction
// ─── UX friction ───
if (s.errorCount === 0 && (s.rageClickCount > 0 || s.deadClickCount > 0)) {
tags.push("ux_friction");
severity = bump(severity, "P2");
}
// upgrade_hesitation
if (s.startUrl && /\/(upgrade|subscription)/i.test(s.startUrl) && s.durationMs > 60_000) {
tags.push("upgrade_hesitation");
severity = bump(severity, "P2");
}
// api_key_friction
if (s.startUrl && /\/api-keys/i.test(s.startUrl) && s.durationMs > 60_000) {
tags.push("api_key_friction");
// ─── VIN decode failure pattern ───
const vinFails = events.filter((e) => e.name === "vin_decode_failed");
if (vinFails.length >= 2) {
const providers = new Set(
vinFails.map((e) => String(e.properties.provider_attempted ?? e.properties.source ?? "")),
);
if (providers.size === 1 && [...providers][0]) {
tags.push("vin_decode_fail_pattern");
severity = bump(severity, "P1");
} else {
// Different providers failing → still notable
tags.push("vin_decode_repeated_failure");
severity = bump(severity, "P2");
}
} else if (vinFails.length === 1) {
// Single failure is still a quality signal, less severe
tags.push("vin_decode_failed_single");
severity = bump(severity, "P3");
}
// onboarding_stuck: requires Sase RO lookup (caller provides ctx)
if (ctx && ctx.signupRecent && !ctx.firstVinQuery && s.durationMs > 120_000) {
if (has(events, "provider_fallback_triggered")) {
tags.push("provider_reliability_issue");
severity = bump(severity, "P2");
}
// ─── Payment criticality ───
const paymentStarted = has(events, "payment_initiated");
const paymentSucceeded = has(events, "payment_succeeded"); // canonical (covers v1 payment_success too)
const paymentSucceededUi = has(events, "payment_success_ui");
const paymentFailed = has(events, "payment_failed");
if (paymentSucceeded && !paymentSucceededUi) {
// Only flag if Sase.tr is on v2 (has payment_success_ui at all) — otherwise this is normal pre-migration state.
const v2InUse = events.some((e) => e.name === "payment_success_ui" || e.rawName === "payment_success_ui");
if (v2InUse) {
tags.push("payment_ui_silent_failure");
severity = bump(severity, "P0");
}
}
if (paymentStarted && !paymentSucceeded && !paymentFailed) {
tags.push("payment_friction");
severity = bump(severity, "P1");
}
if (paymentFailed) {
tags.push("payment_failed_session");
severity = bump(severity, "P1");
}
// ─── Conversion funnel ───
if (has(events, "checkout_started") && !paymentSucceeded) {
tags.push("checkout_abandonment");
severity = bump(severity, "P2");
}
if (has(events, "downgrade_offer_shown")) {
const decided =
has(events, "downgrade_offer_accepted") || has(events, "downgrade_offer_declined");
if (!decided) {
tags.push("downgrade_pending");
severity = bump(severity, "P2");
}
}
// ─── Parts / compatibility ───
const compatChecks = events.filter((e) => e.name === "compatibility_check_completed");
if (compatChecks.some((e) => ["unknown", "incompatible"].includes(String(e.properties.result ?? "")))) {
tags.push("compatibility_quality_gap");
severity = bump(severity, "P2");
}
if (has(events, "parts_export_initiated") && !has(events, "parts_export_completed")) {
tags.push("parts_export_abandoned");
severity = bump(severity, "P3");
}
// ─── Search friction ───
if (count(events, "search_input_validation_failed") >= 3) {
tags.push("search_validation_friction");
severity = bump(severity, "P3");
}
// ─── Onboarding stuck (PostHog identify cache replaces Sase RO lookup) ───
const daysSinceSignup = pick<number>(userProps, "days_since_signup");
if (
typeof daysSinceSignup === "number" &&
daysSinceSignup < 1 &&
!has(events, "vin_decode_succeeded") &&
s.durationMs > 120_000
) {
tags.push("onboarding_stuck");
severity = bump(severity, "P2");
}
// power_user_path (rough heuristic — many clicks, auth, low error)
if (s.isAuthenticated && s.clickCount >= 20 && s.errorCount === 0 && tags.length === 0) {
tags.push("power_user_path");
// ─── Upgrade hesitation (rrweb + custom event combo) ───
if (
s.startUrl &&
/\/(upgrade|subscription|pricing)/i.test(s.startUrl) &&
!has(events, "plan_selected")
) {
if (s.durationMs > 60_000) {
tags.push("upgrade_hesitation");
severity = bump(severity, "P2");
}
}
return { tags, severity };
// ─── Power user path ───
const vinSuccesses = events.filter((e) => e.name === "vin_decode_succeeded");
if (vinSuccesses.length >= 10) {
const providers = new Set(
vinSuccesses.map((e) => String(e.properties.provider ?? "")).filter(Boolean),
);
if (providers.size >= 2 && has(events, "parts_export_initiated")) {
tags.push("power_user_path");
// severity stays INFO
}
}
// ─── At-risk active session (group analytics + user props) ───
const subStatus = pick<string>(userProps, "subscription_status");
const queries30d = pick<number>(userProps, "total_vin_queries_last_30d");
if (subStatus === "active" && typeof queries30d === "number" && queries30d === 0) {
tags.push("at_risk_active_session");
severity = bump(severity, "P2");
}
// ─── Provider mismatch ───
if (has(events, "provider_fallback_triggered") && has(events, "vin_decode_succeeded")) {
// User had to fall back to find a working provider for the same VIN
tags.push("provider_mismatch");
severity = bump(severity, "P2");
}
// ─── API key friction (rrweb URL + custom event combo) ───
if (s.startUrl && /\/api-keys/i.test(s.startUrl) && s.durationMs > 60_000) {
if (!has(events, "api_key_created")) {
tags.push("api_key_friction");
severity = bump(severity, "P3");
}
}
// ─── Webhook setup struggle ───
if (
s.startUrl &&
/\/webhooks?/i.test(s.startUrl) &&
!has(events, "webhook_configured") &&
s.durationMs > 60_000
) {
tags.push("webhook_setup_struggle");
severity = bump(severity, "P3");
}
// Group-context priority bump (B2B high-value)
if (groupProps) {
const mrr = pick<number>(groupProps, "mrr_usd");
if (typeof mrr === "number" && mrr >= 100) {
// bump payment_friction / checkout_abandonment to P1 if high-MRR account
for (const t of ["payment_friction", "checkout_abandonment"]) {
if (tags.includes(t)) severity = bump(severity, "P1");
}
}
}
return { tags: Array.from(new Set(tags)), severity };
}
export function scoreSession(
s: SessionMeta,
sev: SeverityLevel,
novelty: number, // 0..1
patternStrength: number, // 0..1
novelty: number,
patternStrength: number,
groupProps: Record<string, unknown> | null = null,
): number {
const severityWeight: Record<SeverityLevel, number> = {
P0: 1.0,
@@ -82,7 +240,7 @@ export function scoreSession(
INFO: 0.1,
};
const userValue = !s.isAuthenticated
let userValue = !s.isAuthenticated
? 0.2
: s.subscriptionTier?.toLowerCase().includes("full")
? 1.0
@@ -90,6 +248,13 @@ export function scoreSession(
? 0.7
: 0.5;
// Group-context boost
if (groupProps) {
const mrr = (groupProps.mrr_usd ?? 0) as number;
if (mrr >= 200) userValue = Math.max(userValue, 1.0);
else if (mrr >= 100) userValue = Math.max(userValue, 0.8);
}
const ageHours = (Date.now() - s.startedAt.getTime()) / 3_600_000;
const recency = ageHours < 1 ? 1.0 : ageHours < 24 ? 0.7 : ageHours < 168 ? 0.4 : 0.2;