clover's git

the-knife

1	import {
2	MergebotConfiguration,
3	MergebotDeployments,
4	MergebotOpenPullRequestResult,
5	MergebotOperationError,
6	MergebotPullRequestCheck,
7	type MergebotDeployProductionInput,
8	type MergebotOpenPullRequestInput,
9	MergebotPullRequestDetails,
10	MergebotPullRequestLiveStatus,
11	MergebotPullRequestStatus,
12	type MergebotPullRequestInput,
13	type MergebotQueueInput,
14	MergebotRepoQueue,
15	type MergebotSetAutoMergeInput,
16	type MergebotUpdatePullRequestInput,
17	PullRequestActor,
18	} from "@t3tools/contracts";
19	import * as GitHubCredentials from "@t3tools/source-control-github/server/GitHubCredentials";
20	import * as HostProcess from "@t3tools/shared/HostProcess";
21	import * as NodeSocket from "@effect/platform-node/NodeSocket";
22	import * as Struct from "effect/Struct";
23	import * as Clock from "effect/Clock";
24	import * as Context from "effect/Context";
25	import * as Effect from "effect/Effect";
26	import * as FileSystem from "effect/FileSystem";
27	import * as Layer from "effect/Layer";
28	import * as Path from "effect/Path";
29	import * as PubSub from "effect/PubSub";
30	import * as Redacted from "effect/Redacted";
31	import * as Schedule from "effect/Schedule";
32	import * as Schema from "effect/Schema";
33	import * as Stream from "effect/Stream";
34	import * as Socket from "effect/socket/Socket";
35	import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/http";
36	import { deployments, pullRequestLiveStatus, PullRequestRecord, RepoPayload } from "./overview.ts";
37
38	const Comment = Schema.Struct({
39	id: Schema.String,
40	kind: Schema.Literals(["comment", "review", "review_comment"]),
41	authorId: Schema.NullOr(Schema.String),
42	body: Schema.String,
43	url: Schema.String,
44	createdAt: Schema.String,
45	updatedAt: Schema.String,
46	state: Schema.NullOr(Schema.String),
47	path: Schema.NullOr(Schema.String),
48	});
49	const RepoChannel = Schema.Struct({ kind: Schema.Literal("repo_overview"), repoId: Schema.String });
50	const DetailChannel = Schema.Struct({
51	kind: Schema.Literal("pull_request_detail"),
52	repoId: Schema.String,
53	prNumber: Schema.Number,
54	});
55	const Message = Schema.Union([
56	Schema.Struct({
57	type: Schema.Literal("snapshot"),
58	channelKey: Schema.String,
59	channel: RepoChannel,
60	observedAt: Schema.NullOr(Schema.String),
61	payload: RepoPayload,
62	}),
63	Schema.Struct({
64	type: Schema.Literal("snapshot"),
65	channelKey: Schema.String,
66	channel: DetailChannel,
67	observedAt: Schema.NullOr(Schema.String),
68	payload: Schema.Struct({
69	entities: Schema.Struct({
70	pullRequests: Schema.Record(Schema.String, PullRequestRecord),
71	users: Schema.Record(Schema.String, PullRequestActor),
72	}),
73	view: Schema.Struct({
74	repoId: Schema.String,
75	prId: Schema.String,
76	isLoadingDetails: Schema.Boolean,
77	checks: Schema.Array(MergebotPullRequestCheck),
78	comments: Schema.Array(Comment),
79	}),
80	}),
81	}),
82	Schema.Struct({
83	type: Schema.Literal("error"),
84	channelKey: Schema.optionalKey(Schema.String),
85	message: Schema.String,
86	}),
87	]);
88	const decodeMessage = Schema.decodeEffect(Schema.fromJsonString(Message));
89	const encodeStatus = Schema.encodeSync(MergebotPullRequestStatus);
90	const queueEquivalence = Schema.toEquivalence(Schema.NullOr(MergebotRepoQueue));
91	const liveStatusEquivalence = Schema.toEquivalence(Schema.NullOr(MergebotPullRequestLiveStatus));
92	const deploymentsEquivalence = Schema.toEquivalence(Schema.NullOr(MergebotDeployments));
93	const decodeApiError = Schema.decodeUnknownEffect(Schema.Struct({ message: Schema.String }));
94	const decodeOpenedPullRequest = Schema.decodeUnknownEffect(
95	Schema.Struct({ number: MergebotOpenPullRequestResult.fields.number }),
96	);
97	const decodeSession = Schema.decodeUnknownEffect(
98	Schema.Struct({
99	token: Schema.String.check(Schema.isMinLength(1)),
100	}),
101	);
102
103	class MergebotAuthenticationError extends Schema.TaggedError<MergebotAuthenticationError>()(
104	"MergebotAuthenticationError",
105	{ status: Schema.Number },
106	) {}
107
108	type Channel = typeof RepoChannel.Type | typeof DetailChannel.Type;
109	type Change = { readonly repository: string; readonly number?: number };
110
111	export class MergebotQueue extends Context.Service<
112	MergebotQueue,
113	{
114	readonly configuration: MergebotConfiguration | null;
115	readonly read: (input: MergebotQueueInput) => Effect.Effect<MergebotRepoQueue | null>;
116	readonly watchQueue: (input: MergebotQueueInput) => Stream.Stream<MergebotRepoQueue | null>;
117	readonly readPullRequest: (
118	input: MergebotPullRequestInput,
119	) => Effect.Effect<MergebotPullRequestDetails | null>;
120	readonly findPullRequest: (input: {
121	readonly repository: string;
122	readonly headBranch: string;
123	}) => Effect.Effect<MergebotPullRequestDetails | null>;
124	readonly watchPullRequestStatus: (
125	input: MergebotPullRequestInput,
126	) => Stream.Stream<MergebotPullRequestLiveStatus | null>;
127	readonly watchDeployments: (
128	input: MergebotQueueInput,
129	) => Stream.Stream<MergebotDeployments | null>;
130	readonly openPullRequest: (
131	input: MergebotOpenPullRequestInput,
132	) => Effect.Effect<MergebotOpenPullRequestResult, MergebotOperationError>;
133	readonly setAutoMerge: (
134	input: MergebotSetAutoMergeInput,
135	) => Effect.Effect<void, MergebotOperationError>;
136	readonly updatePullRequest: (
137	input: MergebotUpdatePullRequestInput,
138	) => Effect.Effect<void, MergebotOperationError>;
139	readonly deployProduction: (
140	input: MergebotDeployProductionInput,
141	) => Effect.Effect<void, MergebotOperationError>;
142	readonly streamChanges: Stream.Stream<Change>;
143	}
144	>()("t3/mergebot/MergebotQueue") {}
145
146	const encodeSubscribe = Schema.encodeSync(
147	Schema.fromJsonString(
148	Schema.Struct({
149	type: Schema.Literal("subscribe"),
150	channel: Schema.Union([RepoChannel, DetailChannel]),
151	}),
152	),
153	);
154
155	const encodeUnsubscribe = Schema.encodeSync(
156	Schema.fromJsonString(
157	Schema.Struct({
158	type: Schema.Literal("unsubscribe"),
159	channelKey: Schema.String,
160	}),
161	),
162	);
163
164	const channelKey = (channel: Channel) =>
165	channel.kind === "repo_overview"
166	? `repo_overview:${channel.repoId}`
167	: `pull_request_detail:${channel.repoId}#${channel.prNumber}`;
168
169	const make = Effect.gen(function* () {
170	const fs = yield* FileSystem.FileSystem;
171	const path = yield* Path.Path;
172	const home = yield* HostProcess.HomeDirectory;
173	const configuration = yield* fs.readFileString(path.join(home, ".config/mergebot.json")).pipe(
174	Effect.flatMap(Schema.decodeUnknownEffect(Schema.fromJsonString(MergebotConfiguration))),
175	Effect.orElseSucceed(() => null),
176	);
177	const http = HttpClient.withScope(yield* HttpClient.HttpClient);
178	const credentials = yield* GitHubCredentials.GitHubCredentials;
179	const constructSocket = yield* Socket.WebSocketConstructor;
180	const scope = yield* Effect.scope;
181	const changes = yield* PubSub.unbounded<Change>();
182	const repos = new Map<string, RepoPayload | null>();
183	const details = new Map<string, MergebotPullRequestDetails | null>();
184	const channels = new Map<string, Channel>();
185	const observations = new Map<string, number>();
186	const retryAt = new Map<string, number>();
187	const lastReadAt = new Map<string, number>();
188	let writer: Socket.Writer | undefined;
189	let started = false;
190	let account: string | undefined;
191	let activeSocket: Socket.WebSocketLike | undefined;
192	let session: { readonly fingerprint: string; readonly token: Redacted.Redacted } | undefined;
193	/** Live status watchers per detail channel; a watched channel is never idle. */
194	const watchers = new Map<string, number>();
195
196	const repositoryFor = (repository: string) =>
197	configuration?.repositories.find((repo) => repo.toLowerCase() === repository.toLowerCase());
198
199	const invalidate = Effect.fnUntraced(function* (key?: string) {
200	const invalidated: Change[] = [];
201	for (const [currentKey, channel] of channels) {
202	if (key !== undefined && currentKey !== key) continue;
203	const previous =
204	channel.kind === "repo_overview" ? repos.get(channel.repoId) : details.get(currentKey);
205	const changed = previous !== null || observations.has(currentKey);
206	observations.delete(currentKey);
207	retryAt.delete(currentKey);
208	if (channel.kind === "repo_overview") {
209	repos.set(channel.repoId, null);
210	} else details.set(currentKey, null);
211	if (changed)
212	invalidated.push({
213	repository: channel.repoId,
214	...(channel.kind === "pull_request_detail" ? { number: channel.prNumber } : {}),
215	});
216	}
217	yield* PubSub.publishAll(changes, invalidated);
218	});
219
220	const queueFor = (repository: string): MergebotRepoQueue | null | undefined => {
221	const payload = repos.get(repository);
222	if (payload === null || payload === undefined) return payload;
223	if (!configuration) return null;
224	const url = `${configuration.origin}/${repository}`;
225	return {
226	repo: repository,
227	url,
228	status: "ready",
229	processing: payload.entities.repos[repository]!.processing,
230	pullRequests: Object.values(payload.entities.pullRequests).map((pr) => {
231	const position = payload.view.queuePrIds.indexOf(pr.id);
232	return encodeStatus({
233	...pr,
234	position: position < 0 ? null : position + 1,
235	url: `${url}/pull/${pr.number}`,
236	});
237	}),
238	};
239	};
240
241	const consume = Effect.fnUntraced(function* (text: string) {
242	const message = yield* decodeMessage(text);
243	if (message.type === "error") {
244	yield* invalidate(message.channelKey);
245	const nextAttempt = (yield* Clock.currentTimeMillis) + 1_000;
246	for (const key of channels.keys()) {
247	if (message.channelKey === undefined || message.channelKey === key)
248	retryAt.set(key, nextAttempt);
249	}
250	return;
251	}
252	const channel = channels.get(message.channelKey);
253	if (!channel || channelKey(message.channel) !== message.channelKey) return;
254	const repository = channel.repoId;
255	if (!configuration) return;
256	const url = `${configuration.origin}/${repository}`;
257	const observedAt = message.observedAt === null ? NaN : Date.parse(message.observedAt);
258	if (Number.isFinite(observedAt)) {
259	const previous = observations.get(message.channelKey);
260	if (previous !== undefined && observedAt < previous) return;
261	observations.set(message.channelKey, observedAt);
262	}
263	const entities = message.payload.entities;
264	if ("repos" in entities) {
265	const payload = message.payload;
266	const view = payload.view;
267	if (!("queuePrIds" in view) || view.repoId !== repository) {
268	yield* invalidate(message.channelKey);
269	return;
270	}
271	if (!message.observedAt || !Number.isFinite(Date.parse(message.observedAt))) {
272	repos.delete(repository);
273	yield* PubSub.publish(changes, { repository });
274	return;
275	}
276	const repo = entities.repos[repository];
277	if (!repo) {
278	yield* invalidate(message.channelKey);
279	return;
280	}
281	const pullRequestRecords = Object.values(entities.pullRequests).filter(
282	(pr) => pr.repoId === repository && pr.id === `${repository}#${pr.number}`,
283	);
284	if (pullRequestRecords.length !== Object.keys(entities.pullRequests).length) {
285	yield* invalidate(message.channelKey);
286	return;
287	}
288	repos.set(repository, { entities, view });
289	for (const record of pullRequestRecords) {
290	const key = channelKey({
291	kind: "pull_request_detail",
292	repoId: repository,
293	prNumber: record.number,
294	});
295	const detail = details.get(key);
296	if (
297	detail &&
298	(detail.headSha !== record.headSha || detail.lifecycle !== record.lifecycle) &&
299	(observations.get(key) ?? -Infinity) <= observedAt
300	) {
301	details.delete(key);
302	}
303	}
304	yield* PubSub.publish(changes, { repository });
305	return;
306	}
307	if (message.channel.kind !== "pull_request_detail" || !("users" in entities)) return;
308	const id = `${repository}#${message.channel.prNumber}`;
309	const pr = message.payload.entities.pullRequests[id];
310	const view = message.payload.view;
311	if (!("prId" in view) || view.repoId !== repository || view.prId !== id) {
312	yield* invalidate(message.channelKey);
313	return;
314	}
315	if (
316	!message.observedAt ||
317	!Number.isFinite(Date.parse(message.observedAt)) ||
318	view.isLoadingDetails
319	) {
320	details.delete(message.channelKey);
321	} else if (
322	pr?.id === id &&
323	pr.repoId === repository &&
324	pr.number === message.channel.prNumber
325	) {
326	const known = repos.get(repository)?.entities.pullRequests[id];
327	if (
328	known &&
329	(known.headSha !== pr.headSha || known.lifecycle !== pr.lifecycle) &&
330	observedAt < (observations.get(`repo_overview:${repository}`) ?? -Infinity)
331	) {
332	details.delete(message.channelKey);
333	yield* PubSub.publish(changes, { repository, number: pr.number });
334	return;
335	}
336	const queue = queueFor(repository);
337	const queued = queue?.pullRequests?.find((entry) => entry.number === pr.number);
338	details.set(message.channelKey, {
339	...Struct.omit(pr, ["id", "repoId", "authorId"]),
340	author: pr.authorId === null ? null : (entities.users[pr.authorId] ?? null),
341	observedAt,
342	comments: view.comments.map((comment) => ({
343	id: comment.id,
344	kind:
345	comment.kind === "comment"
346	? "issue-comment"
347	: comment.kind === "review_comment"
348	? "review-comment"
349	: "review",
350	author: comment.authorId === null ? null : (entities.users[comment.authorId] ?? null),
351	body: comment.body,
352	createdAt: comment.createdAt,
353	editedAt: comment.updatedAt,
354	url: comment.url,
355	path: comment.path,
356	reviewState: comment.state,
357	})),
358	position: queued && queued.headSha === pr.headSha ? queued.position : null,
359	url: `${url}/pull/${pr.number}`,
360	checks: view.checks,
361	});
362	} else {
363	details.set(message.channelKey, null);
364	}
365	yield* PubSub.publish(changes, { repository, number: message.channel.prNumber });
366	});
367
368	const exchange = Effect.fnUntraced(function* (
369	origin: string,
370	credential: GitHubCredentials.GitHubCredential,
371	) {
372	const response = yield* http
373	.execute(
374	HttpClientRequest.post(`${origin}/api/auth/github/token`).pipe(
375	HttpClientRequest.setHeader(
376	"Authorization",
377	`Bearer ${Redacted.value(credential.token)}`,
378	),
379	HttpClientRequest.setHeader("accept", "application/json"),
380	),
381	)
382	.pipe(
383	Effect.provideService(FetchHttpClient.RequestInit, { redirect: "manual" }),
384	Effect.timeout("10 seconds"),
385	);
386	if (response.status !== 200) {
387	if (response.status === 401) yield* credentials.invalidate("github.com");
388	return yield* new MergebotAuthenticationError({ status: response.status });
389	}
390	const { token } = yield* response.json.pipe(Effect.flatMap(decodeSession));
391	session = { fingerprint: credential.fingerprint, token: Redacted.make(token) };
392	return session.token;
393	});
394
395	const connect = Effect.gen(function* () {
396	if (!configuration) return;
397	const credential = yield* credentials.get("github.com");
398	if (account !== undefined && account !== credential.fingerprint) {
399	yield* invalidate();
400	}
401	account = credential.fingerprint;
402	const token = yield* exchange(configuration.origin, credential);
403	const socket = yield* Socket.fromWebSocket(
404	Effect.acquireRelease(
405	Effect.sync(() => {
406	activeSocket = constructSocket(
407	`${configuration.origin.replace(/^https:/, "wss:")}/api/state/ws`,
408	{ headers: { Authorization: `Bearer ${Redacted.value(token)}` } },
409	);
410	return activeSocket;
411	}),
412	(socket) => Effect.sync(() => socket.close(1000)),
413	),
414	{ openTimeout: "10 seconds" },
415	);
416	return yield* Effect.scoped(
417	Effect.gen(function* () {
418	const { pull } = yield* socket.reader;
419	const currentWriter = yield* socket.writer;
420	writer = currentWriter;
421	for (const channel of channels.values()) {
422	yield* currentWriter.write(encodeSubscribe({ type: "subscribe", channel }));
423	}
424	while (true) {
425	const frames = yield* pull;
426	for (const frame of frames) {
427	yield* consume(typeof frame === "string" ? frame : new TextDecoder().decode(frame));
428	}
429	}
430	}),
431	).pipe(
432	Effect.ensuring(
433	Effect.sync(() => {
434	writer = undefined;
435	activeSocket = undefined;
436	}),
437	),
438	);
439	});
440
441	const authorize = Effect.fnUntraced(function* () {
442	if (activeSocket && activeSocket.readyState >= 2) {
443	yield* invalidate();
444	return null;
445	}
446	const credential = yield* credentials.get("github.com").pipe(Effect.orElseSucceed(() => null));
447	if (credential && (account === undefined || credential.fingerprint === account))
448	return credential.fingerprint;
449	yield* invalidate();
450	activeSocket?.close(1000);
451	return null;
452	});
453
454	const releaseIdleChannels = Effect.gen(function* () {
455	const now = yield* Clock.currentTimeMillis;
456	for (const [key, at] of lastReadAt) {
457	if (watchers.has(key) || now - at < 5 * 60_000) continue;
458	lastReadAt.delete(key);
459	channels.delete(key);
460	details.delete(key);
461	observations.delete(key);
462	retryAt.delete(key);
463	if (writer)
464	yield* writer
465	.write(encodeUnsubscribe({ type: "unsubscribe", channelKey: key }))
466	.pipe(Effect.orElseSucceed(() => undefined));
467	}
468	});
469
470	const start = Effect.fnUntraced(function* () {
471	if (started || !configuration) return;
472	started = true;
473	for (const repository of configuration.repositories) {
474	const channel = { kind: "repo_overview", repoId: repository } as const;
475	channels.set(channelKey(channel), channel);
476	}
477	yield* Effect.forkIn(
478	releaseIdleChannels.pipe(Effect.repeat(Schedule.spaced("1 minute"))),
479	scope,
480	);
481	yield* Effect.forkIn(
482	connect.pipe(
483	Effect.scoped,
484	Effect.ensuring(invalidate()),
485	Effect.retry({ schedule: Schedule.spaced("1 second") }),
486	),
487	scope,
488	);
489	});
490
491	const retryChannel = Effect.fnUntraced(function* (key: string) {
492	const nextAttempt = retryAt.get(key);
493	if (nextAttempt === undefined || !writer || (yield* Clock.currentTimeMillis) < nextAttempt)
494	return;
495	const channel = channels.get(key);
496	if (!channel) return;
497	retryAt.delete(key);
498	if (channel.kind === "repo_overview") repos.delete(channel.repoId);
499	else details.delete(key);
500	yield* writer
501	.write(encodeSubscribe({ type: "subscribe", channel }))
502	.pipe(Effect.catch(() => Effect.sync(() => activeSocket?.close(1000))));
503	});
504
505	const waitFor = <A>(fingerprint: string, read: () => A | null | undefined) =>
506	Effect.gen(function* () {
507	const value = yield* Effect.gen(function* () {
508	const subscription = yield* PubSub.subscribe(changes);
509	let value = read();
510	while (value === undefined) {
511	yield* PubSub.take(subscription);
512	value = read();
513	}
514	return value;
515	}).pipe(
516	Effect.scoped,
517	Effect.timeout("10 seconds"),
518	Effect.orElseSucceed(() => null),
519	);
520	return (yield* authorize()) === fingerprint ? value : null;
521	});
522
523	const subscribeDetail = Effect.fnUntraced(function* (repository: string, number: number) {
524	const channel = { kind: "pull_request_detail", repoId: repository, prNumber: number } as const;
525	const key = channelKey(channel);
526	if (!channels.has(key)) {
527	channels.set(key, channel);
528	if (writer)
529	yield* writer
530	.write(encodeSubscribe({ type: "subscribe", channel }))
531	.pipe(Effect.orElseSucceed(() => undefined));
532	}
533	yield* retryChannel(key);
534	return key;
535	});
536
537	/** Resolves a configured repository for the current account and keeps its overview live. */
538	const openOverview = Effect.fnUntraced(function* (input: string) {
539	const repository = repositoryFor(input);
540	if (!repository) return null;
541	const fingerprint = yield* authorize();
542	if (fingerprint === null) return null;
543	yield* start();
544	yield* retryChannel(channelKey({ kind: "repo_overview", repoId: repository }));
545	return { repository, fingerprint };
546	});
547
548	const readPullRequest = Effect.fn("MergebotQueue.readPullRequest")(function* (
549	input: MergebotPullRequestInput,
550	) {
551	const repository = repositoryFor(input.repository);
552	if (!repository) return null;
553	const fingerprint = yield* authorize();
554	if (fingerprint === null) return null;
555	yield* start();
556	const key = yield* subscribeDetail(repository, input.number);
557	lastReadAt.set(key, yield* Clock.currentTimeMillis);
558	return yield* waitFor(fingerprint, () => details.get(key));
559	});
560
561	const read = Effect.fn("MergebotQueue.read")(function* (input: MergebotQueueInput) {
562	const opened = yield* openOverview(input.repository);
563	if (!opened) return null;
564	return yield* waitFor(opened.fingerprint, () => queueFor(opened.repository));
565	});
566
567	const readPullRequestStatus = Effect.fn("MergebotQueue.readPullRequestStatus")(function* (
568	input: MergebotPullRequestInput,
569	) {
570	const opened = yield* openOverview(input.repository);
571	if (!opened || !configuration) return null;
572	const { origin } = configuration;
573	const { repository } = opened;
574	const key = yield* subscribeDetail(repository, input.number);
575	return yield* waitFor(opened.fingerprint, () => {
576	const payload = repos.get(repository);
577	const detail = details.get(key);
578	if (!payload) return payload;
579	// Closed and long-merged PRs exist only in their detail snapshot.
580	const inOverview = payload.entities.pullRequests[`${repository}#${input.number}`];
581	if (!inOverview && detail === undefined) return undefined;
582	return pullRequestLiveStatus(
583	`${origin}/${repository}`,
584	repository,
585	payload,
586	detail,
587	input.number,
588	);
589	});
590	});
591
592	const readDeployments = Effect.fn("MergebotQueue.readDeployments")(function* (
593	input: MergebotQueueInput,
594	) {
595	const opened = yield* openOverview(input.repository);
596	if (!opened || !configuration) return null;
597	const { origin } = configuration;
598	const { repository } = opened;
599	return yield* waitFor(opened.fingerprint, () => {
600	const payload = repos.get(repository);
601	return payload && deployments(`${origin}/${repository}`, repository, payload);
602	});
603	});
604
605	const watch = <A>(
606	repository: string,
607	read: Effect.Effect<A>,
608	equivalence: (self: A, that: A) => boolean,
609	) =>
610	Stream.unwrap(
611	Effect.gen(function* () {
612	const subscription = yield* PubSub.subscribe(changes);
613	return Stream.concat(
614	Stream.fromEffect(read),
615	Stream.fromSubscription(subscription).pipe(
616	Stream.filter((change) => change.repository.toLowerCase() === repository.toLowerCase()),
617	Stream.mapEffect(() => read),
618	),
619	);
620	}),
621	).pipe(Stream.changesWith(equivalence));
622
623	/** Sends one mutation with the Mergebot session, exchanging a fresh session once on 401. */
624	const request = Effect.fnUntraced(function* (
625	operation: string,
626	input: { readonly repository: string },
627	method: "POST" | "PATCH",
628	path: string,
629	body: object,
630	) {
631	const repository = repositoryFor(input.repository);
632	if (!configuration || !repository)
633	return yield* new MergebotOperationError({
634	operation,
635	status: null,
636	detail: `${input.repository} is not a Mergebot repository`,
637	});
638	const { origin } = configuration;
639
640	const sessionToken = Effect.gen(function* () {
641	const credential = yield* credentials.get("github.com").pipe(
642	Effect.mapError(
643	() =>
644	new MergebotOperationError({
645	operation,
646	status: null,
647	detail: "GitHub is not signed in",
648	}),
649	),
650	);
651	return session?.fingerprint === credential.fingerprint
652	? session.token
653	: yield* exchange(origin, credential);
654	});
655	const send = (token: Redacted.Redacted) =>
656	http
657	.execute(
658	HttpClientRequest.make(method)(`${origin}/api/repos/${repository}${path}`).pipe(
659	HttpClientRequest.setHeader("Authorization", `Bearer ${Redacted.value(token)}`),
660	HttpClientRequest.setHeader("accept", "application/json"),
661	HttpClientRequest.bodyJsonUnsafe(body),
662	),
663	)
664	.pipe(Effect.timeout("30 seconds"));
665
666	const response = yield* Effect.gen(function* () {
667	const first = yield* send(yield* sessionToken);
668	if (first.status !== 401) return first;
669	session = undefined;
670	return yield* send(yield* sessionToken);
671	}).pipe(
672	Effect.mapError((cause) =>
673	cause._tag === "MergebotOperationError"
674	? cause
675	: cause._tag === "MergebotAuthenticationError"
676	? new MergebotOperationError({
677	operation,
678	status: cause.status,
679	detail: "Mergebot did not accept the GitHub sign-in",
680	})
681	: new MergebotOperationError({
682	operation,
683	status: null,
684	detail: "Mergebot is unreachable",
685	}),
686	),
687	);
688	const json = yield* response.json.pipe(Effect.orElseSucceed(() => undefined));
689	if (response.status >= 200 && response.status < 300)
690	return { json, url: `${origin}/${repository}` };
691
692	const detail = yield* decodeApiError(json).pipe(
693	Effect.map((error) => error.message),
694	Effect.orElseSucceed(() => `HTTP ${response.status}`),
695	);
696	return yield* new MergebotOperationError({ operation, status: response.status, detail });
697	}, Effect.scoped);
698
699	const setAutoMerge = Effect.fn("MergebotQueue.setAutoMerge")(function* (
700	input: MergebotSetAutoMergeInput,
701	) {
702	yield* request("update auto-merge", input, "POST", `/pulls/${input.number}/auto-merge`, {
703	enabled: input.enabled,
704	});
705	});
706
707	return MergebotQueue.of({
708	configuration,
709	read,
710	watchQueue: (input) => watch(input.repository, read(input), queueEquivalence),
711	readPullRequest,
712	findPullRequest: Effect.fn("MergebotQueue.findPullRequest")(function* (input) {
713	const opened = yield* openOverview(input.repository);
714	if (!opened) return null;
715	const { repository } = opened;
716	const queue = yield* waitFor(opened.fingerprint, () => queueFor(repository));
717	if (!queue || queue.status !== "ready") return null;
718	const matches = Object.values(repos.get(repository)?.entities.pullRequests ?? {})
719	.filter((pr) => pr.headBranch === input.headBranch)
720	.sort(
721	(a, b) =>
722	Number(b.lifecycle === "open") - Number(a.lifecycle === "open") ||
723	b.updatedAt.localeCompare(a.updatedAt),
724	);
725	const match = matches?.[0];
726	return match ? yield* readPullRequest({ repository, number: match.number }) : null;
727	}),
728	watchPullRequestStatus: (input) =>
729	Stream.unwrap(
730	Effect.gen(function* () {
731	const repository = repositoryFor(input.repository);
732	if (repository) {
733	const key = channelKey({
734	kind: "pull_request_detail",
735	repoId: repository,
736	prNumber: input.number,
737	});
738	yield* Effect.acquireRelease(
739	Effect.sync(() => watchers.set(key, (watchers.get(key) ?? 0) + 1)),
740	() =>
741	Effect.map(Clock.currentTimeMillis, (now) => {
742	const remaining = (watchers.get(key) ?? 1) - 1;
743	if (remaining > 0) {
744	watchers.set(key, remaining);
745	} else {
746	// The last watcher hands the channel to the idle lease.
747	watchers.delete(key);
748	lastReadAt.set(key, now);
749	}
750	}),
751	);
752	}
753	return watch(input.repository, readPullRequestStatus(input), liveStatusEquivalence);
754	}),
755	),
756	watchDeployments: (input) =>
757	watch(input.repository, readDeployments(input), deploymentsEquivalence),
758	openPullRequest: Effect.fn("MergebotQueue.openPullRequest")(function* (input) {
759	const operation = "open the pull request";
760	const pullRequest = Struct.omit(input, ["repository", "autoMerge"]);
761	const created = yield* request(operation, input, "POST", "/pulls", pullRequest);
762	const { number } = yield* decodeOpenedPullRequest(created.json).pipe(
763	Effect.mapError(
764	() =>
765	new MergebotOperationError({
766	operation,
767	status: null,
768	detail: "Mergebot returned an unexpected response",
769	}),
770	),
771	);
772	// The PR exists now, so failing here would invite a retry that collides with it.
773	if (input.autoMerge)
774	yield* setAutoMerge({ repository: input.repository, number, enabled: true }).pipe(
775	Effect.catchTags({
776	MergebotOperationError: (error) =>
777	Effect.logWarning("Mergebot opened the pull request without auto-merge", {
778	status: error.status,
779	}),
780	}),
781	);
782	return { number, url: `${created.url}/pull/${number}` };
783	}),
784	setAutoMerge,
785	updatePullRequest: Effect.fn("MergebotQueue.updatePullRequest")(function* (input) {
786	const changes = Struct.omit(input, ["repository", "number"]);
787	yield* request("update the pull request", input, "PATCH", `/pulls/${input.number}`, changes);
788	}),
789	deployProduction: Effect.fn("MergebotQueue.deployProduction")(function* (input) {
790	yield* request("deploy to production", input, "POST", "/deploy/production", {
791	sha: input.sha,
792	});
793	}),
794	streamChanges: Stream.fromPubSub(changes),
795	});
796	});
797
798	export const layerWithSocketConstructor = Layer.effect(MergebotQueue, make);
799	export const layer = layerWithSocketConstructor.pipe(
800	Layer.provide(NodeSocket.layerWebSocketConstructorWS),
801	);