Blame · Line-by-line history

`merge-queue.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.

merge-queue.tsBlame313 lines · 1 contributor

Normal view Raw

a79a9edClaude	1	/**
	2	* Block E5 — Merge queue helpers.
	3	*
	4	* A merge queue serialises merges on `(repository_id, base_branch)`: instead
	5	* of merging a PR immediately, it's enqueued. A worker (or the manual
	6	* "process next" button surfaced on the queue UI) pops the head of the queue,
	7	* re-runs gates against the latest base, and — if green — performs the merge.
	8	*
	9	* This module is deliberately minimal: no side-effects on gate execution or
	10	* the actual git merge (those are owned by `pulls.tsx`). We just manage
	11	* the queue state + ordering. Every DB path is wrapped to never throw.
	12	*/
	13
	14	import { and, asc, eq, sql } from "drizzle-orm";
	15	import { db } from "../db";
	16	import { mergeQueueEntries, pullRequests } from "../db/schema";
	17	import type { MergeQueueEntry } from "../db/schema";
	18
	19	export interface EnqueueArgs {
	20	repositoryId: string;
	21	pullRequestId: string;
	22	baseBranch: string;
	23	enqueuedBy?: string \| null;
	24	}
	25
	26	export interface EnqueueResult {
	27	ok: boolean;
	28	entry?: MergeQueueEntry;
	29	reason?: string;
	30	}
	31
	32	/**
	33	* Append a PR to the end of the queue for its `(repo, baseBranch)`. No-op
	34	* (returns ok:false with a reason) if the PR is already queued or running.
	35	*/
	36	export async function enqueuePr(args: EnqueueArgs): Promise<EnqueueResult> {
	37	try {
	38	// Check for existing active entry for this PR.
	39	const existing = await db
	40	.select()
	41	.from(mergeQueueEntries)
	42	.where(eq(mergeQueueEntries.pullRequestId, args.pullRequestId));
	43	const active = existing.find(
	44	(e) => e.state === "queued" \|\| e.state === "running"
	45	);
	46	if (active) {
	47	return { ok: false, reason: "Pull request is already in the queue." };
	48	}
	49
	50	// Compute next position in this (repo, base) queue.
	51	const rows = await db
	52	.select({ maxPos: sql<number>`COALESCE(MAX(${mergeQueueEntries.position}), -1)` })
	53	.from(mergeQueueEntries)
	54	.where(
	55	and(
	56	eq(mergeQueueEntries.repositoryId, args.repositoryId),
	57	eq(mergeQueueEntries.baseBranch, args.baseBranch),
	58	sql`${mergeQueueEntries.state} IN ('queued','running')`
	59	)
	60	);
	61	const nextPos = (rows[0]?.maxPos ?? -1) + 1;
	62
	63	const [entry] = await db
	64	.insert(mergeQueueEntries)
	65	.values({
	66	repositoryId: args.repositoryId,
	67	pullRequestId: args.pullRequestId,
	68	baseBranch: args.baseBranch,
	69	position: nextPos,
	70	enqueuedBy: args.enqueuedBy \|\| null,
	71	state: "queued",
	72	})
	73	.returning();
	74	return { ok: true, entry };
	75	} catch (err) {
	76	console.error("[merge-queue] enqueue:", err);
	77	return { ok: false, reason: "Failed to enqueue pull request." };
	78	}
	79	}
	80
	81	/**
	82	* Remove an active entry from the queue (user-initiated cancel, or a PR
	83	* closed while queued). Marks it `dequeued` rather than deleting for audit.
	84	*/
	85	export async function dequeueEntry(entryId: string): Promise<boolean> {
	86	try {
	87	const res = await db
	88	.update(mergeQueueEntries)
	89	.set({ state: "dequeued", finishedAt: new Date() })
	90	.where(
	91	and(
	92	eq(mergeQueueEntries.id, entryId),
	93	sql`${mergeQueueEntries.state} IN ('queued','running')`
	94	)
	95	)
	96	.returning({ id: mergeQueueEntries.id });
	97	return res.length > 0;
	98	} catch (err) {
	99	console.error("[merge-queue] dequeue:", err);
	100	return false;
	101	}
	102	}
	103
	104	/**
	105	* Peek the head of the queue for a `(repo, baseBranch)` pair. Returns the
	106	* oldest `queued` entry — the one that would be popped next by processNext.
	107	*/
	108	export async function peekHead(
	109	repositoryId: string,
	110	baseBranch: string
	111	): Promise<MergeQueueEntry \| null> {
	112	try {
	113	const rows = await db
	114	.select()
	115	.from(mergeQueueEntries)
	116	.where(
	117	and(
	118	eq(mergeQueueEntries.repositoryId, repositoryId),
	119	eq(mergeQueueEntries.baseBranch, baseBranch),
	120	eq(mergeQueueEntries.state, "queued")
	121	)
	122	)
	123	.orderBy(asc(mergeQueueEntries.position), asc(mergeQueueEntries.enqueuedAt))
	124	.limit(1);
	125	return rows[0] \|\| null;
	126	} catch {
	127	return null;
	128	}
	129	}
	130
	131	/**
	132	* List queue entries for a repo, newest-first per base branch. Includes
	133	* terminal states so the queue UI can show recent merges/failures.
	134	*/
	135	export async function listQueue(
	136	repositoryId: string,
	137	opts: { limit?: number; baseBranch?: string } = {}
	138	): Promise<MergeQueueEntry[]> {
	139	const limit = opts.limit ?? 100;
	140	try {
	141	if (opts.baseBranch) {
	142	return await db
	143	.select()
	144	.from(mergeQueueEntries)
	145	.where(
	146	and(
	147	eq(mergeQueueEntries.repositoryId, repositoryId),
	148	eq(mergeQueueEntries.baseBranch, opts.baseBranch)
	149	)
	150	)
	151	.orderBy(asc(mergeQueueEntries.position), asc(mergeQueueEntries.enqueuedAt))
	152	.limit(limit);
	153	}
	154	return await db
	155	.select()
	156	.from(mergeQueueEntries)
	157	.where(eq(mergeQueueEntries.repositoryId, repositoryId))
	158	.orderBy(asc(mergeQueueEntries.position), asc(mergeQueueEntries.enqueuedAt))
	159	.limit(limit);
	160	} catch {
	161	return [];
	162	}
	163	}
	164
	165	/**
	166	* Transition the head entry → `running`. Returns the entry (if any) so the
	167	* caller can kick off gates + perform the merge. The caller must eventually
	168	* call `completeEntry` with success/failure.
	169	*/
	170	export async function markHeadRunning(
	171	repositoryId: string,
	172	baseBranch: string
	173	): Promise<MergeQueueEntry \| null> {
	174	const head = await peekHead(repositoryId, baseBranch);
	175	if (!head) return null;
	176	try {
	177	const [updated] = await db
	178	.update(mergeQueueEntries)
	179	.set({ state: "running", startedAt: new Date() })
	180	.where(
	181	and(
	182	eq(mergeQueueEntries.id, head.id),
	183	eq(mergeQueueEntries.state, "queued")
	184	)
	185	)
	186	.returning();
	187	return updated \|\| null;
	188	} catch {
	189	return null;
	190	}
	191	}
	192
	193	/**
	194	* Mark a running entry as finished. `state` is the final state
	195	* (`merged` \| `failed`). Non-running entries are left untouched.
	196	*/
	197	export async function completeEntry(
	198	entryId: string,
	199	finalState: "merged" \| "failed",
	200	errorMessage?: string
	201	): Promise<boolean> {
	202	try {
	203	const res = await db
	204	.update(mergeQueueEntries)
	205	.set({
	206	state: finalState,
	207	finishedAt: new Date(),
	208	errorMessage: errorMessage \|\| null,
	209	})
	210	.where(eq(mergeQueueEntries.id, entryId))
	211	.returning({ id: mergeQueueEntries.id });
	212	return res.length > 0;
	213	} catch (err) {
	214	console.error("[merge-queue] complete:", err);
	215	return false;
	216	}
	217	}
	218
	219	/**
	220	* Is this PR currently queued or running? Convenience helper for the merge
	221	* UI (so we can swap the button label to "In queue…").
	222	*/
	223	export async function isQueued(pullRequestId: string): Promise<boolean> {
	224	try {
	225	const rows = await db
	226	.select({ id: mergeQueueEntries.id })
	227	.from(mergeQueueEntries)
	228	.where(
	229	and(
	230	eq(mergeQueueEntries.pullRequestId, pullRequestId),
	231	sql`${mergeQueueEntries.state} IN ('queued','running')`
	232	)
	233	)
	234	.limit(1);
	235	return rows.length > 0;
	236	} catch {
	237	return false;
	238	}
	239	}
	240
	241	/**
	242	* Check queue depth for `(repo, baseBranch)` — number of `queued` + `running`.
	243	*/
	244	export async function queueDepth(
	245	repositoryId: string,
	246	baseBranch: string
	247	): Promise<number> {
	248	try {
	249	const rows = await db
	250	.select({ n: sql<number>`COUNT(*)` })
	251	.from(mergeQueueEntries)
	252	.where(
	253	and(
	254	eq(mergeQueueEntries.repositoryId, repositoryId),
	255	eq(mergeQueueEntries.baseBranch, baseBranch),
	256	sql`${mergeQueueEntries.state} IN ('queued','running')`
	257	)
	258	);
	259	return Number(rows[0]?.n \|\| 0);
	260	} catch {
	261	return 0;
	262	}
	263	}
	264
	265	/**
	266	* Resolve PR metadata (number, title) for a list of entries — the queue UI
	267	* needs those to render links. Kept in the helper so routes don't have to
	268	* re-join.
	269	*/
	270	export interface QueueEntryWithPr extends MergeQueueEntry {
	271	prNumber: number \| null;
	272	prTitle: string \| null;
	273	prState: string \| null;
	274	prHeadBranch: string \| null;
	275	prAuthorId: string \| null;
	276	}
	277
	278	export async function listQueueWithPrs(
	279	repositoryId: string
	280	): Promise<QueueEntryWithPr[]> {
	281	try {
	282	const rows = await db
	283	.select({
	284	id: mergeQueueEntries.id,
	285	repositoryId: mergeQueueEntries.repositoryId,
	286	pullRequestId: mergeQueueEntries.pullRequestId,
	287	baseBranch: mergeQueueEntries.baseBranch,
	288	state: mergeQueueEntries.state,
	289	position: mergeQueueEntries.position,
	290	enqueuedBy: mergeQueueEntries.enqueuedBy,
	291	enqueuedAt: mergeQueueEntries.enqueuedAt,
	292	startedAt: mergeQueueEntries.startedAt,
	293	finishedAt: mergeQueueEntries.finishedAt,
	294	errorMessage: mergeQueueEntries.errorMessage,
	295	prNumber: pullRequests.number,
	296	prTitle: pullRequests.title,
	297	prState: pullRequests.state,
	298	prHeadBranch: pullRequests.headBranch,
	299	prAuthorId: pullRequests.authorId,
	300	})
	301	.from(mergeQueueEntries)
	302	.leftJoin(
	303	pullRequests,
	304	eq(mergeQueueEntries.pullRequestId, pullRequests.id)
	305	)
	306	.where(eq(mergeQueueEntries.repositoryId, repositoryId))
	307	.orderBy(asc(mergeQueueEntries.position), asc(mergeQueueEntries.enqueuedAt))
	308	.limit(200);
	309	return rows as QueueEntryWithPr[];
	310	} catch {
	311	return [];
	312	}
	313	}