feat(replays): store rrweb chunk payloads in R2, not ClickHouse

apps/api/alchemy.run.ts

@@ -136,6 +136,33 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) =>
 			name: planetScaleWebhookQueueName,
 		})
 
+		// Session-replay rrweb payloads. The ingest gateway (a Railway container,
+		// not a Worker) writes these over the S3 API with SigV4; this binding is
+		// the read side, hydrating `session_replay_events` rows whose `Events` is
+		// empty. Stage-isolated, so a pr/stg deploy can never serve or overwrite
+		// prd recordings.
+		//
+		// The 32-day expiry is deliberately LONGER than the table's 30-day TTL:
+		// the row must disappear before the object does. The other way round
+		// leaves a session that lists as recorded but plays back empty, which is
+		// the one failure mode with no good client-side handling.
+		const replayBlobs = yield* Cloudflare.R2.Bucket("replay-blobs", {
+			name: resolveWorkerName("replay-blobs", stage),
+			// Deliberately unprefixed, so the rule covers whatever key scheme is
+			// current. `replay_object_key` is versioned (`v1/…`) precisely so a
+			// format change can write under a new prefix while the old one ages
+			// out — a rule pinned to `v1/` would silently stop expiring anything
+			// the moment that happens, and the bucket would grow forever with no
+			// failing test to catch it. Nothing else writes here.
+			lifecycleRules: [
+				{
+					id: "expire-replay-chunks",
+					enabled: true,
+					deleteObjectsTransition: { condition: { type: "Age", maxAge: 32 * 24 * 60 * 60 } },
+				},
+			],
+		})
+
 		const worker = (yield* Cloudflare.Worker("api", {
 			name: resolveWorkerName("api", stage),
 			main: path.join(import.meta.dirname, "src", "worker.ts"),
@@ -167,6 +194,9 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) =>
 				AI: Cloudflare.AI.Gateway("maple-api-ai"),
 				CHAT_SESSION: chatSession,
 				MCP_SESSIONS: mcpSessions,
+				// Read side of the replay payload store; absent bindings degrade to
+				// inline-only hydration (see platform/ReplayBlobStore.ts).
+				REPLAY_BLOBS: replayBlobs,
 				VCS_SYNC_QUEUE: vcsSyncQueue,
 				VCS_SYNC_QUEUE_NAME: vcsSyncQueueName,
 				PLANETSCALE_WEBHOOK_QUEUE: planetScaleWebhookQueue,

apps/api/src/platform/ReplayBlobStore.test.ts

@@ -0,0 +1,137 @@
+import { describe, expect, it } from "vitest"
+import { Effect, Layer } from "effect"
+import { layerFromEnvRecord } from "@maple/effect-cloudflare/worker-environment"
+import { ReplayBlobStore, replayObjectKey, REPLAY_BLOBS_BINDING } from "./ReplayBlobStore"
+
+const gzip = async (text: string): Promise<Uint8Array> => {
+	const stream = new Blob([text]).stream().pipeThrough(new CompressionStream("gzip"))
+	return new Uint8Array(await new Response(stream).arrayBuffer())
+}
+
+/** Minimal stand-in for the R2 binding: only `get` is exercised. */
+const fakeBucket = (objects: Record<string, Uint8Array>, onGet?: (key: string) => void) => ({
+	get: async (key: string) => {
+		onGet?.(key)
+		const bytes = objects[key]
+		if (!bytes) return null
+		return { bytes: async () => bytes }
+	},
+})
+
+const runWithBucket = <A>(
+	bucket: unknown,
+	program: (store: typeof ReplayBlobStore.Service) => Effect.Effect<A>,
+) =>
+	Effect.runPromise(
+		Effect.gen(function* () {
+			const store = yield* ReplayBlobStore
+			return yield* program(store)
+		}).pipe(
+			Effect.provide(
+				ReplayBlobStore.layer.pipe(
+					Layer.provide(layerFromEnvRecord({ [REPLAY_BLOBS_BINDING]: bucket })),
+				),
+			),
+		),
+	)
+
+describe("replayObjectKey", () => {
+	// These expectations are duplicated verbatim in `replay_object_key`'s tests
+	// in apps/ingest/src/r2.rs. Nothing at runtime reconciles the two — the
+	// writer and the reader agree only by construction — and a divergence reads
+	// to a user as "every recording is empty", not as an error. So both suites
+	// pin the same strings.
+	it("matches the ingest gateway's key scheme", () => {
+		expect(replayObjectKey("org_123", "sess_abc", 7)).toBe("v1/org_123/sess_abc/00000007.json.gz")
+	})
+
+	it("zero-pads so lexicographic order is playback order", () => {
+		const keys = [
+			replayObjectKey("o", "s", 10),
+			replayObjectKey("o", "s", 2),
+			replayObjectKey("o", "s", 1),
+		].sort()
+		expect(keys).toEqual([
+			"v1/o/s/00000001.json.gz",
+			"v1/o/s/00000002.json.gz",
+			"v1/o/s/00000010.json.gz",
+		])
+	})
+
+	it("does not truncate a chunk seq wider than the pad", () => {
+		expect(replayObjectKey("o", "s", 123_456_789)).toBe("v1/o/s/123456789.json.gz")
+	})
+})
+
+describe("ReplayBlobStore.hydrate", () => {
+	const events = '[{"type":2,"timestamp":1}]'
+
+	it("fills in payloads for blob-backed chunks", async () => {
+		const bucket = fakeBucket({
+			"v1/org_1/sess_1/00000000.json.gz": await gzip(events),
+		})
+		const result = await runWithBucket(bucket, (store) =>
+			store.hydrate("org_1", "sess_1", [{ chunkSeq: 0, events: "", byteSize: 26 }]),
+		)
+		expect(result).toEqual([{ chunkSeq: 0, events, byteSize: 26 }])
+	})
+
+	it("leaves pre-cutover chunks untouched and never fetches them", async () => {
+		// The dual-read. A row written before the R2 cutover carries its payload
+		// inline; going to the bucket for it would 404 and silently drop a chunk
+		// that was there all along.
+		const fetched: string[] = []
+		const bucket = fakeBucket({}, (key) => fetched.push(key))
+		const result = await runWithBucket(bucket, (store) =>
+			store.hydrate("org_1", "sess_1", [{ chunkSeq: 0, events }]),
+		)
+		expect(result).toEqual([{ chunkSeq: 0, events }])
+		expect(fetched).toEqual([])
+	})
+
+	it("drops a chunk whose object is missing rather than failing the request", async () => {
+		const bucket = fakeBucket({
+			"v1/org_1/sess_1/00000001.json.gz": await gzip(events),
+		})
+		const result = await runWithBucket(bucket, (store) =>
+			store.hydrate("org_1", "sess_1", [
+				{ chunkSeq: 0, events: "" },
+				{ chunkSeq: 1, events: "" },
+			]),
+		)
+		expect(result).toEqual([{ chunkSeq: 1, events }])
+	})
+
+	it("preserves chunk order despite concurrent fetches", async () => {
+		const objects: Record<string, Uint8Array> = {}
+		for (let seq = 0; seq < 20; seq++) {
+			objects[replayObjectKey("org_1", "sess_1", seq)] = await gzip(`[${seq}]`)
+		}
+		const result = await runWithBucket(objects && fakeBucket(objects), (store) =>
+			store.hydrate(
+				"org_1",
+				"sess_1",
+				Array.from({ length: 20 }, (_, seq) => ({ chunkSeq: seq, events: "" })),
+			),
+		)
+		expect(result.map((chunk) => chunk.events)).toEqual(
+			Array.from({ length: 20 }, (_, seq) => `[${seq}]`),
+		)
+	})
+
+	it("is a no-op when the binding is absent", async () => {
+		// Self-hosted and the Docker image of this API have no R2 at all. Every
+		// row there carries its payload inline, so hydration must pass through
+		// rather than erroring on a missing binding.
+		const chunks = [{ chunkSeq: 0, events }]
+		const result = await Effect.runPromise(
+			Effect.gen(function* () {
+				const store = yield* ReplayBlobStore
+				return yield* store.hydrate("org_1", "sess_1", chunks)
+			}).pipe(
+				Effect.provide(ReplayBlobStore.layer.pipe(Layer.provide(layerFromEnvRecord({})))),
+			),
+		)
+		expect(result).toEqual(chunks)
+	})
+})

apps/api/src/platform/ReplayBlobStore.ts

@@ -0,0 +1,136 @@
+import { Context, Effect, Layer, Option } from "effect"
+import { R2Bucket, type R2BucketClient } from "@maple/effect-cloudflare/r2-bucket"
+import { WorkerEnvironment } from "@/platform/WorkerEnvironment"
+
+// ---------------------------------------------------------------------------
+// ReplayBlobStore — reads rrweb chunk payloads out of R2.
+//
+// The ingest gateway stores each replay chunk's gzipped rrweb array as an
+// object and writes a `session_replay_events` row that carries only the chunk's
+// metadata, with an empty `Events`. This service turns those rows back into the
+// inline payloads the player already expects, so the HTTP response shape is
+// unchanged.
+//
+// Two states are normal, not errors:
+//
+//   - **No binding.** Self-hosted installs and the Docker image of this API have
+//     no R2. `WorkerEnvironment` resolves to `{}` outside a Worker isolate, so
+//     `bindOptional` yields `None` and hydration becomes a no-op.
+//   - **Non-empty `events` on a row.** That row predates the R2 cutover (or came
+//     from a BYO-ClickHouse org, which never stops writing inline). It passes
+//     through untouched. This is the dual-read, and it is the whole migration
+//     strategy: no backfill, the old rows age out on the table's 30-day TTL.
+// ---------------------------------------------------------------------------
+
+/** Binding name; must match the key in `apps/api/alchemy.run.ts`. */
+export const REPLAY_BLOBS_BINDING = "REPLAY_BLOBS"
+
+/**
+ * Object key for one replay chunk.
+ *
+ * **Must stay byte-identical to `replay_object_key` in `apps/ingest/src/r2.rs`.**
+ * Nothing at runtime cross-checks the two — a divergence reads as "every
+ * recording is empty", so the shared shape is asserted in both test suites.
+ */
+export const replayObjectKey = (orgId: string, sessionId: string, chunkSeq: number): string =>
+	`v1/${orgId}/${sessionId}/${String(chunkSeq).padStart(8, "0")}.json.gz`
+
+/** The subset of a `session_replay_events` row hydration needs. */
+export interface HydratableChunk {
+	readonly chunkSeq: number
+	readonly events: string
+}
+
+export interface ReplayBlobStoreShape {
+	/**
+	 * Fill in `events` for every blob-backed chunk, preserving order.
+	 *
+	 * Chunks whose object is missing are **dropped**, not failed: a single
+	 * absent chunk should cost you that slice of the recording, not the whole
+	 * session. This matches how the player already tolerates a chunk it can't
+	 * parse.
+	 */
+	readonly hydrate: <T extends HydratableChunk>(
+		orgId: string,
+		sessionId: string,
+		chunks: readonly T[],
+	) => Effect.Effect<T[]>
+}
+
+// Chunk fetches are independent; the player needs all of them before it can
+// build the Replayer, so latency is the max, not the sum — but only if they
+// overlap. Sequential gets across a long session would be a visible regression
+// against the single ClickHouse round trip this replaces.
+const FETCH_CONCURRENCY = 8
+
+const decodeGzip = (bytes: Uint8Array): Promise<string> =>
+	new Response(
+		new Blob([bytes as unknown as BlobPart]).stream().pipeThrough(new DecompressionStream("gzip")),
+	).text()
+
+const makeHydrate =
+	(bucket: R2BucketClient) =>
+	<T extends HydratableChunk>(orgId: string, sessionId: string, chunks: readonly T[]) =>
+		Effect.forEach(
+			chunks,
+			(chunk) => {
+				// Pre-cutover row: the payload is already in hand.
+				if (chunk.events !== "") return Effect.succeed(Option.some(chunk))
+				const key = replayObjectKey(orgId, sessionId, chunk.chunkSeq)
+				return bucket.get(key).pipe(
+					Effect.flatMap((object) =>
+						object === null
+							? Effect.succeed(Option.none<T>())
+							: object
+									.bytes()
+									.pipe(
+										Effect.flatMap((bytes) =>
+											Effect.promise(() => decodeGzip(bytes)),
+										),
+										Effect.map((events) => Option.some({ ...chunk, events })),
+									),
+					),
+					// A failed fetch degrades the recording rather than the request.
+					// Logged at warning because a nonzero rate here means either the
+					// bucket lifecycle is outrunning the table TTL or the ingest-side
+					// key scheme has drifted from this one.
+					Effect.catch((cause) =>
+						Effect.logWarning("replay chunk payload unavailable").pipe(
+							Effect.annotateLogs({ orgId, sessionId, key, cause: String(cause) }),
+							Effect.as(Option.none<T>()),
+						),
+					),
+				)
+			},
+			{ concurrency: FETCH_CONCURRENCY },
+		).pipe(Effect.map((results) => results.filter(Option.isSome).map((result) => result.value)))
+
+export class ReplayBlobStore extends Context.Service<ReplayBlobStore, ReplayBlobStoreShape>()(
+	"@maple/api/platform/ReplayBlobStore",
+	{
+		make: Effect.gen(function* () {
+			// The R2 client's methods each demand `WorkerEnvironment` again at call
+			// time, so capture it here and close over it. Otherwise the requirement
+			// leaks into every caller's type, and the routes would have to know
+			// they're on Cloudflare.
+			const env = yield* WorkerEnvironment
+			const bucket = yield* R2Bucket.bindOptional(R2Bucket(REPLAY_BLOBS_BINDING))
+			if (Option.isNone(bucket)) {
+				// Expected on self-hosted / Docker / tests. Every row will carry its
+				// payload inline there, so hydration has nothing to do.
+				return { hydrate: (_orgId, _sessionId, chunks) => Effect.succeed([...chunks]) }
+			}
+			const hydrate = makeHydrate(bucket.value)
+			return {
+				hydrate: (orgId, sessionId, chunks) =>
+					hydrate(orgId, sessionId, chunks).pipe(
+						Effect.provideService(WorkerEnvironment, env),
+					),
+			}
+		}),
+	},
+) {
+	static readonly layer = Layer.effect(this, this.make)
+}
+
+export const ReplayBlobStoreLive = ReplayBlobStore.layer.pipe(Layer.provide(WorkerEnvironment.layer))

apps/api/src/routes/v1/session-replay.http.ts

@@ -13,17 +13,19 @@ import {
 	TraceId,
 	UserId,
 } from "@maple/domain/http"
-import { Effect, Option, Schema } from "effect"
+import { Effect, Layer, Option, Schema } from "effect"
 import { CH } from "@maple/query-engine"
 import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService"
+import { ReplayBlobStore, ReplayBlobStoreLive } from "@/platform/ReplayBlobStore"
 
 const decodeSessionId = Schema.decodeSync(SessionId)
 const decodeTraceId = Schema.decodeSync(TraceId)
 const decodeUserId = Schema.decodeSync(UserId)
 
-export const HttpSessionReplaysLive = HttpApiBuilder.group(MapleApi, "sessionReplays", (handlers) =>
+const HttpSessionReplaysGroup = HttpApiBuilder.group(MapleApi, "sessionReplays", (handlers) =>
 	Effect.gen(function* () {
 		const warehouse = yield* WarehouseQueryService
+		const blobs = yield* ReplayBlobStore
 
 		return handlers
 			.handle("listReplays", ({ payload }) =>
@@ -195,12 +197,15 @@ export const HttpSessionReplaysLive = HttpApiBuilder.group(MapleApi, "sessionRep
 							sessionId: payload.sessionId,
 						},
 					)
-					const chunks = yield* warehouse.compiledQuery(tenant, compiled, {
+					const rows = yield* warehouse.compiledQuery(tenant, compiled, {
 						profile: "list",
 						context: "getReplayEvents",
 					})
-					// rrweb payloads come straight from ClickHouse (no R2 / presigning);
-					// each chunk's `events` is the rrweb array JSON the player parses.
+					// Each chunk's `events` is the rrweb array JSON the player parses.
+					// Rows written since the R2 cutover carry an empty `events` and the
+					// payload comes from the blob store; older rows carry it inline and
+					// pass through untouched.
+					const chunks = yield* blobs.hydrate(tenant.orgId, payload.sessionId, rows)
 					return new GetReplayEventsResponse({ chunks })
 				}),
 			)
@@ -294,3 +299,10 @@ export const HttpSessionReplaysLive = HttpApiBuilder.group(MapleApi, "sessionRep
 			)
 	}),
 )
+
+// Self-contained: the blob store resolves from the worker env (and degrades to
+// a no-op hydrator when the R2 binding is absent), so it is provided here
+// rather than pushed onto every caller that builds this group — the v2 route
+// tests construct their own layer stack and would otherwise have to know about
+// a storage detail of one handler.
+export const HttpSessionReplaysLive = HttpSessionReplaysGroup.pipe(Layer.provide(ReplayBlobStoreLive))