Blame · Line-by-line history

`pr-live.ts`

Each line is annotated with the commit that last touched it. Click any SHA to jump to that commit and see the surrounding change.

pr-live.tsBlame364 lines · 1 contributor

Normal view Raw

3c03977Claude	1	/**
	2	* PR live co-editing — presence, cursors, and content patches.
	3	*
	4	* One row per active browser tab (human) or agent runtime in
	5	* `pr_live_sessions`. The HTTP layer (src/routes/pr-live.ts) opens a
	6	* topic-based SSE stream `pr-live:<prId>`; this lib drives the writes:
	7	*
	8	* joinSession — register a session, return id + colour
	9	* updateCursor — debounce-friendly cursor broadcast
	10	* broadcastEdit — content patch fan-out (last-write-wins for v1)
	11	* leaveSession — explicit "this tab is closing" marker
	12	* sweepStale — autopilot task: idle/left state machine
	13	*
	14	* Everything that talks to subscribers funnels through `publish()` on
	15	* topic `pr-live:<prId>` so the route file stays a thin transport. DB
	16	* writes are best-effort — if the database is offline the broadcast
	17	* still reaches the in-process subscribers, which is the happy path
	18	* for production where ephemeral presence rarely needs durability.
	19	*/
	20
	21	import { eq, and, lt, sql, ne } from "drizzle-orm";
	22	import { createHash } from "crypto";
	23	import { db } from "../db";
	24	import { prLiveSessions } from "../db/schema";
	25	import { publish } from "./sse";
	26
	27	/** Field identifiers the client may attach a cursor / patch to. */
	28	export type LiveField =
	29	\| "description"
	30	\| `comment_${string}`
	31	\| `line_${string}:${number}`;
	32
	33	/** Cursor range — character offsets within the named field's content. */
	34	export interface CursorPosition {
	35	field: LiveField \| string;
	36	range: { start: number; end: number };
	37	}
	38
	39	/** Patch payload — opaque to the lib; the client picks the dialect. */
	40	export interface EditPatch {
	41	field: LiveField \| string;
	42	/** Op shape — replace, insert, delete. */
	43	op: "replace" \| "insert" \| "delete";
	44	/** Position (char offset) the op applies at. */
	45	at: number;
	46	/** Inserted/replaced text (omit for delete). */
	47	value?: string;
	48	/** Length to remove (replace/delete only). */
	49	length?: number;
	50	}
	51
	52	/** Public session view — the shape SSE consumers see. */
	53	export interface PublicLiveSession {
	54	id: string;
	55	prId: string;
	56	userId: string \| null;
	57	agentSessionId: string \| null;
	58	color: string;
	59	status: "active" \| "idle" \| "left";
	60	cursor: CursorPosition \| null;
	61	joinedAt: string;
	62	lastSeenAt: string;
	63	}
	64
	65	/** Tuned per the spec: idle after 60s, dropped after 5m of silence. */
	66	export const IDLE_AFTER_MS = 60_000;
	67	export const LEFT_AFTER_MS = 5 * 60_000;
	68	/** Heartbeat the client is expected to send. */
	69	export const HEARTBEAT_MS = 15_000;
	70
	71	/** Deterministic colour palette — picked by hashing the principal id. */
	72	const COLORS = [
	73	"#f87171", "#fb923c", "#fbbf24", "#a3e635",
	74	"#34d399", "#22d3ee", "#60a5fa", "#a78bfa",
	75	"#f472b6", "#fb7185",
	76	] as const;
	77
	78	/**
	79	* Deterministic colour for a principal (user id or agent session id).
	80	* Hash → modulo palette so concurrent tabs of the same user share the
	81	* same hue and the UI stays steady.
	82	*/
	83	export function colorFor(principalId: string): string {
	84	const h = createHash("sha256").update(principalId).digest();
	85	const idx = h[0] % COLORS.length;
	86	return COLORS[idx];
	87	}
	88
	89	/** SSE topic name for one PR. */
	90	export function prLiveTopic(prId: string): string {
	91	return `pr-live:${prId}`;
	92	}
	93
	94	/**
	95	* Best-effort DB write. If the DB is unreachable we still want presence
	96	* to work in-process — production already runs autopilot cleanup that
	97	* heals from drift. Returns the value or `null` on error.
	98	*/
	99	async function tryDb<T>(fn: () => Promise<T>): Promise<T \| null> {
	100	try {
	101	return await fn();
	102	} catch {
	103	return null;
	104	}
	105	}
	106
	107	/**
	108	* Begin a session. Exactly one of `userId` / `agentSessionId` should be
	109	* non-null; the DB stores both as nullable so a single table represents
	110	* both principals.
	111	*/
	112	export async function joinSession(args: {
	113	prId: string;
	114	userId?: string \| null;
	115	agentSessionId?: string \| null;
	116	}): Promise<{ sessionId: string; color: string } \| null> {
	117	const principalId =
	118	args.userId ?? args.agentSessionId ?? null;
	119	if (!principalId) return null;
	120	const color = colorFor(principalId);
	121
	122	const row = await tryDb(async () => {
	123	const [inserted] = await db
	124	.insert(prLiveSessions)
	125	.values({
	126	prId: args.prId,
	127	userId: args.userId ?? null,
	128	agentSessionId: args.agentSessionId ?? null,
	129	color,
	130	status: "active",
	131	})
	132	.returning();
	133	return inserted ?? null;
	134	});
	135
	136	// Even if the DB write failed, mint a synthetic id so the in-process
	137	// broadcaster has something to key on for this tab.
	138	const sessionId = row?.id ?? `tmp-${principalId}-${Date.now()}`;
	139
	140	publish(prLiveTopic(args.prId), {
	141	event: "presence-join",
	142	data: {
	143	sessionId,
	144	prId: args.prId,
	145	userId: args.userId ?? null,
	146	agentSessionId: args.agentSessionId ?? null,
	147	color,
	148	status: "active",
	149	joinedAt: row?.joinedAt?.toISOString() ?? new Date().toISOString(),
	150	},
	151	});
	152
	153	return { sessionId, color };
	154	}
	155
	156	/**
	157	* Update cursor position + touch heartbeat. Caller is expected to
	158	* debounce (~100ms) — this function does not internally throttle.
	159	*/
	160	export async function updateCursor(
	161	sessionId: string,
	162	prId: string,
	163	position: CursorPosition
	164	): Promise<void> {
	165	await tryDb(async () => {
	166	await db
	167	.update(prLiveSessions)
	168	.set({
	169	cursorPosition: position as unknown as Record<string, unknown>,
	170	lastSeenAt: new Date(),
	171	status: "active",
	172	})
	173	.where(eq(prLiveSessions.id, sessionId));
	174	});
	175
	176	publish(prLiveTopic(prId), {
	177	event: "cursor",
	178	data: { sessionId, position },
	179	});
	180	}
	181
	182	/**
	183	* Touch a session's heartbeat without changing the cursor. The client
	184	* pings every 15s; missing pings transition the row to idle/left.
	185	*/
	186	export async function heartbeat(
	187	sessionId: string,
	188	prId: string
	189	): Promise<void> {
	190	await tryDb(async () => {
	191	await db
	192	.update(prLiveSessions)
	193	.set({ lastSeenAt: new Date(), status: "active" })
	194	.where(eq(prLiveSessions.id, sessionId));
	195	});
	196
	197	publish(prLiveTopic(prId), {
	198	event: "heartbeat",
	199	data: { sessionId },
	200	});
	201	}
	202
	203	/**
	204	* Broadcast a content patch. v1 is last-write-wins — the client applies
	205	* received patches directly to its textarea. A proper OT engine is the
	206	* v2 follow-up, which is why the patch shape is intentionally open.
	207	*/
	208	export async function broadcastEdit(
	209	sessionId: string,
	210	prId: string,
	211	patch: EditPatch
	212	): Promise<void> {
	213	// Touch last_seen so an editing tab doesn't go stale.
	214	await tryDb(async () => {
	215	await db
	216	.update(prLiveSessions)
	217	.set({ lastSeenAt: new Date(), status: "active" })
	218	.where(eq(prLiveSessions.id, sessionId));
	219	});
	220
	221	publish(prLiveTopic(prId), {
	222	event: "edit",
	223	data: { sessionId, patch },
	224	});
	225	}
	226
	227	/** Mark a session as explicitly left (tab close, navigation, etc.). */
	228	export async function leaveSession(
	229	sessionId: string,
	230	prId: string
	231	): Promise<void> {
	232	await tryDb(async () => {
	233	await db
	234	.update(prLiveSessions)
	235	.set({ status: "left", lastSeenAt: new Date() })
	236	.where(eq(prLiveSessions.id, sessionId));
	237	});
	238
	239	publish(prLiveTopic(prId), {
	240	event: "presence-leave",
	241	data: { sessionId },
	242	});
	243	}
	244
	245	/** List the live presence on a PR (active + idle, drops 'left'). */
	246	export async function listLive(prId: string): Promise<PublicLiveSession[]> {
	247	const rows = await tryDb(() =>
	248	db
	249	.select()
	250	.from(prLiveSessions)
	251	.where(
	252	and(
	253	eq(prLiveSessions.prId, prId),
	254	ne(prLiveSessions.status, "left")
	255	)
	256	)
	257	);
	258	if (!rows) return [];
	259	return rows.map((r) => ({
	260	id: r.id,
	261	prId: r.prId,
	262	userId: r.userId,
	263	agentSessionId: r.agentSessionId,
	264	color: r.color,
	265	status: (r.status as PublicLiveSession["status"]) ?? "active",
	266	cursor: (r.cursorPosition as CursorPosition \| null) ?? null,
	267	joinedAt: r.joinedAt.toISOString(),
	268	lastSeenAt: r.lastSeenAt.toISOString(),
	269	}));
	270	}
	271
	272	/**
	273	* Autopilot sweep — transition stale rows to idle / left based on the
	274	* configured thresholds. Returns { idled, left } counts so the task
	275	* row can log them. Safe to call on an empty DB.
	276	*/
	277	export async function sweepStale(
	278	now: Date = new Date()
	279	): Promise<{ idled: number; left: number }> {
	280	const idleCutoff = new Date(now.getTime() - IDLE_AFTER_MS);
	281	const leftCutoff = new Date(now.getTime() - LEFT_AFTER_MS);
	282
	283	const idledRes = await tryDb(async () => {
	284	const updated = await db
	285	.update(prLiveSessions)
	286	.set({ status: "idle" })
	287	.where(
	288	and(
	289	eq(prLiveSessions.status, "active"),
	290	lt(prLiveSessions.lastSeenAt, idleCutoff)
	291	)
	292	)
	293	.returning({ id: prLiveSessions.id, prId: prLiveSessions.prId });
	294	return updated;
	295	});
	296	const idled = idledRes ?? [];
	297
	298	const leftRes = await tryDb(async () => {
	299	const updated = await db
	300	.update(prLiveSessions)
	301	.set({ status: "left" })
	302	.where(
	303	and(
	304	// Sweeping anything not already 'left' covers both active and
	305	// idle rows that have crossed the 5-minute threshold.
	306	ne(prLiveSessions.status, "left"),
	307	lt(prLiveSessions.lastSeenAt, leftCutoff)
	308	)
	309	)
	310	.returning({ id: prLiveSessions.id, prId: prLiveSessions.prId });
	311	return updated;
	312	});
	313	const left = leftRes ?? [];
	314
	315	// Fire presence-update / leave events for any peers still listening.
	316	for (const row of idled) {
	317	publish(prLiveTopic(row.prId), {
	318	event: "presence-update",
	319	data: { sessionId: row.id, status: "idle" },
	320	});
	321	}
	322	for (const row of left) {
	323	publish(prLiveTopic(row.prId), {
	324	event: "presence-leave",
	325	data: { sessionId: row.id },
	326	});
	327	}
	328
	329	return { idled: idled.length, left: left.length };
	330	}
	331
	332	/**
	333	* Lookup a single session row by id — used by the SSE endpoint to
	334	* verify a heartbeat/leave request matches a real row before
	335	* broadcasting.
	336	*/
	337	export async function getSession(
	338	sessionId: string
	339	): Promise<PublicLiveSession \| null> {
	340	const row = await tryDb(async () => {
	341	const [r] = await db
	342	.select()
	343	.from(prLiveSessions)
	344	.where(eq(prLiveSessions.id, sessionId))
	345	.limit(1);
	346	return r ?? null;
	347	});
	348	if (!row) return null;
	349	return {
	350	id: row.id,
	351	prId: row.prId,
	352	userId: row.userId,
	353	agentSessionId: row.agentSessionId,
	354	color: row.color,
	355	status: (row.status as PublicLiveSession["status"]) ?? "active",
	356	cursor: (row.cursorPosition as CursorPosition \| null) ?? null,
	357	joinedAt: row.joinedAt.toISOString(),
	358	lastSeenAt: row.lastSeenAt.toISOString(),
	359	};
	360	}
	361
	362	// Re-export so callers don't need to also import drizzle helpers. Kept
	363	// at the bottom so the public surface is obvious when reading the file.
	364	export { sql };