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

fix(phase6a): PostHog snapshot blob fetch uses start_blob_key/end_blob_key; parse [window_id,event] tuples

Browse Source 
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Semih
2026-05-13 21:26:36 +00:00
parent 795528982e
commit 93499faeaa
3 changed files with 21 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -22,9 +22,17 @@ export async function runCompressSessions(): Promise<{ compressed: number; faile
try {
const sources = await getSnapshotSources(s.id);
const events: any[] = [];
// Fetch up to first 6 blobs to keep cost bounded
for (const src of sources.slice(0, 6)) {
const blob = await getSnapshotBlob(s.id, src.source, src.blob_key);
// Group blob_keys by source and fetch in a single range request when possible.
const bySource = new Map<string, string[]>();
for (const src of sources) {
if (!bySource.has(src.source)) bySource.set(src.source, []);
bySource.get(src.source)!.push(src.blob_key);
}
for (const [source, keys] of bySource) {
// Sort blob keys numerically (they're stringified ints) and cap at first 12.
const sorted = [...keys].sort((a, b) => Number(a) - Number(b)).slice(0, 12);
if (!sorted.length) continue;
const blob = await getSnapshotBlob(s.id, source, sorted[0], sorted[sorted.length - 1]);
events.push(...parseSnapshotBlob(blob));
}
const out = compressSnapshots(events, {

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

@@ -231,21 +231,18 @@ function normalizeError(msg: string): string {
}
export function parseSnapshotBlob(text: string): RREvent[] {
// PostHog returns ndjson lines. Each line may itself be a JSON
// with `{window_id, data}` envelope where `data` is the rrweb event.
// PostHog returns jsonl where each line is [window_id, event].
const out: RREvent[] = [];
for (const line of text.split("\n")) {
const trimmed = line.trim();
if (!trimmed) continue;
try {
const parsed = JSON.parse(trimmed);
const ev = parsed?.data ?? parsed;
let ev: any = parsed;
if (Array.isArray(parsed) && parsed.length >= 2) ev = parsed[1];
else if (parsed && typeof parsed === "object" && parsed.data && parsed.type === undefined) ev = parsed.data;
if (ev && typeof ev.type === "number" && typeof ev.timestamp === "number") {
out.push(ev as RREvent);
} else if (Array.isArray(ev)) {
for (const e of ev) {
if (e && typeof e.type === "number" && typeof e.timestamp === "number") out.push(e);
}
}
} catch {
// skip

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

@@ -68,17 +68,20 @@ export async function getSnapshotSources(sessionId: string): Promise<PHSnapshotS
return data.sources ?? [];
}
// Returns raw newline-delimited JSON snapshots for a given source/blob.
// Returns raw newline-delimited JSON snapshots for a given source + blob range.
// PostHog requires both start_blob_key and end_blob_key; pass equal values for a single blob.
export async function getSnapshotBlob(
sessionId: string,
source: string,
blobKey: string,
startBlobKey: string,
endBlobKey: string = startBlobKey,
): Promise<string> {
const url = new URL(
`${HOST}/api/projects/${PROJECT_ID}/session_recordings/${sessionId}/snapshots/`,
);
url.searchParams.set("source", source);
url.searchParams.set("blob_key", blobKey);
url.searchParams.set("start_blob_key", startBlobKey);
url.searchParams.set("end_blob_key", endBlobKey);
const res = await fetch(url, { headers: headers() });
if (!res.ok) throw new Error(`posthog blob ${res.status}`);
return await res.text();