1import { describe, expect, it } from "@effect/vitest";
2import * as GitHubCredentials from "@t3tools/source-control-github/server/GitHubCredentials";
3import * as Effect from "effect/Effect";
4import * as FileSystem from "effect/FileSystem";
5import * as HostProcess from "@t3tools/shared/HostProcess";
6import * as NodeServices from "@effect/platform-node/NodeServices";
7import * as Layer from "effect/Layer";
8import * as Redacted from "effect/Redacted";
9import * as Stream from "effect/Stream";
10import * as Fiber from "effect/Fiber";
11import * as Deferred from "effect/Deferred";
12import * as TestClock from "effect/testing/TestClock";
13import * as Socket from "effect/socket/Socket";
14import { HttpClient, HttpClientResponse } from "effect/http";
15import * as MergebotQueue from "./MergebotQueue.ts";
16
17const repository = "work/app";
18const url = `https://mergebot.example/${repository}`;
19const configuration = { origin: "https://mergebot.example", repositories: [repository] };
20const pr = {
21 id: `${repository}#123`,
22 repoId: repository,
23 number: 123,
24 authorId: null,
25 createdAt: "2026-10-14T10:00:00Z",
26 additions: 3,
27 deletions: 1,
28 changedFilesCount: 2,
29 labels: [],
30 title: "Ship change",
31 updatedAt: "2026-10-14T10:02:00Z",
32 body: null,
33 draft: false,
34 headBranch: "feature",
35 baseBranch: "main",
36 headSha: "current-head",
37 lifecycle: "open",
38 queueStatus: "waiting_ci",
39 mergeEtaAt: null,
40 ciStatus: "pending",
41 hasConflicts: false,
42 url: "https://github.com/work/app/pull/123",
43};
44const check = {
45 id: "check:1",
46 name: "Tests",
47 workflowName: "CI",
48 status: "completed",
49 conclusion: "failure",
50 url: null,
51 startedAt: "2026-10-14T10:00:00Z",
52 completedAt: "2026-10-14T10:01:00Z",
53};
54const repo = {
55 processing: true,
56 defaultBranch: "main",
57 productionBranch: "production",
58 stagingUrl: "https://staging.example",
59 productionUrl: "https://app.example",
60 triggerLabel: "merge",
61 stagingDeployment: null,
62 productionDeployment: null,
63};
64const repoChannel = { kind: "repo_overview", repoId: repository };
65const detailChannel = { kind: "pull_request_detail", repoId: repository, prNumber: 123 };
66const repoSnapshot = {
67 observedAt: "2026-10-14T10:02:00Z",
68 type: "snapshot",
69 channel: repoChannel,
70 channelKey: `repo_overview:${repository}`,
71 payload: {
72 entities: {
73 repos: { [repository]: repo },
74 users: {},
75 pullRequests: { [pr.id]: pr },
76 mergedCommits: {},
77 },
78 view: { repoId: repository, queuePrIds: [pr.id], mergedCommitIds: [] },
79 },
80};
81const detailSnapshot = {
82 type: "snapshot",
83 channel: detailChannel,
84 channelKey: `pull_request_detail:${pr.id}`,
85 observedAt: "2026-10-14T10:02:00Z",
86 payload: {
87 entities: { pullRequests: { [pr.id]: pr }, users: {} },
88 view: {
89 repoId: repository,
90 prId: pr.id,
91 comments: [],
92 checks: [check],
93 isLoadingDetails: false,
94 },
95 },
96};
97
98class TestSocket implements Socket.WebSocketLike {
99 readyState = 1;
100 readonly sent: unknown[] = [];
101 readonly listeners = new Map<string, Set<(event: Socket.WebSocketEvent) => void>>();
102 readonly onSubscribe: (socket: TestSocket, channel: { kind: string }) => void;
103 constructor(onSubscribe: (socket: TestSocket, channel: { kind: string }) => void) {
104 this.onSubscribe = onSubscribe;
105 }
106 addEventListener(type: string, listener: (event: Socket.WebSocketEvent) => void) {
107 const set = this.listeners.get(type) ?? new Set();
108 set.add(listener);
109 this.listeners.set(type, set);
110 }
111 removeEventListener(type: string, listener: (event: Socket.WebSocketEvent) => void) {
112 this.listeners.get(type)?.delete(listener);
113 }
114 emit(type: string, event: Socket.WebSocketEvent) {
115 for (const listener of this.listeners.get(type) ?? []) listener(event);
116 }
117 close() {
118 if (this.readyState === 3) return;
119 this.readyState = 3;
120 this.emit("close", { code: 1000, reason: "Test close" });
121 }
122 send(data: string | Uint8Array<ArrayBuffer>) {
123 const message = JSON.parse(String(data));
124 this.sent.push(message);
125 if (message.type === "subscribe") this.onSubscribe(this, message.channel);
126 }
127 snapshot(value: unknown) {
128 this.emit("message", { data: JSON.stringify(value) });
129 }
130}
131
132function harness(configurationText: string | null = JSON.stringify(configuration)) {
133 let token: string | null = "test-github-token";
134 let onSubscribe = (socket: TestSocket, channel: { kind: string }) => {
135 socket.snapshot(channel.kind === "repo_overview" ? repoSnapshot : detailSnapshot);
136 };
137 const sockets: TestSocket[] = [];
138 const exchanges: Array<{ url: string; authorization: string | undefined }> = [];
139 let exchangeStatus = 200;
140 let session = "sealed-mergebot-session";
141 const mutations: Array<{
142 method: string;
143 url: string;
144 authorization: string | undefined;
145 body: unknown;
146 }> = [];
147 let respond = (_request: (typeof mutations)[number]) => Response.json({ ok: true });
148 const requests: Array<{ url: string; options?: Socket.WebSocketConstructorOptions }> = [];
149 return {
150 sockets,
151 requests,
152 exchanges,
153 mutations,
154 respondWith: (handler: typeof respond) => {
155 respond = handler;
156 },
157 setSession: (value: string) => {
158 session = value;
159 },
160 setExchangeStatus: (status: number) => {
161 exchangeStatus = status;
162 },
163 setToken: (value: string | null) => {
164 token = value;
165 },
166 subscribeWith: (handler: typeof onSubscribe) => {
167 onSubscribe = handler;
168 },
169 layer: MergebotQueue.layerWithSocketConstructor.pipe(
170 Layer.provide(
171 Layer.mergeAll(
172 NodeServices.layer,
173 Layer.succeed(
174 HttpClient.HttpClient,
175 HttpClient.make((request) => {
176 if (request.url.endsWith("/api/auth/github/token")) {
177 exchanges.push({ url: request.url, authorization: request.headers.authorization });
178 return Effect.succeed(
179 HttpClientResponse.fromWeb(
180 request,
181 Response.json({ token: session }, { status: exchangeStatus }),
182 ),
183 );
184 }
185 const mutation = {
186 method: request.method,
187 url: request.url,
188 authorization: request.headers.authorization,
189 body:
190 request.body._tag === "Uint8Array"
191 ? JSON.parse(new TextDecoder().decode(request.body.body))
192 : undefined,
193 };
194 mutations.push(mutation);
195 return Effect.succeed(HttpClientResponse.fromWeb(request, respond(mutation)));
196 }),
197 ),
198 Layer.succeed(HostProcess.HomeDirectory, "/test"),
199 Layer.succeed(
200 FileSystem.FileSystem,
201 FileSystem.makeNoop(
202 configurationText === null
203 ? {}
204 : {
205 readFileString: () => Effect.succeed(configurationText),
206 },
207 ),
208 ),
209 Layer.succeed(GitHubCredentials.GitHubCredentials, {
210 get: (host: string) =>
211 token === null
212 ? Effect.fail(new GitHubCredentials.GitHubNotSignedInError({ host }))
213 : Effect.succeed({
214 host,
215 token: Redacted.make(token),
216 fingerprint: token,
217 source: "gh" as const,
218 }),
219 invalidate: () => Effect.void,
220 }),
221 Layer.succeed(Socket.WebSocketConstructor, (url, options) => {
222 requests.push({ url, ...(options ? { options } : {}) });
223 const socket = new TestSocket((...args) => onSubscribe(...args));
224 sockets.push(socket);
225 return socket;
226 }),
227 ),
228 ),
229 ),
230 };
231}
232
233const nextChange = (service: MergebotQueue.MergebotQueue["Service"]) =>
234 Stream.runHead(service.streamChanges).pipe(Effect.forkScoped({ startImmediately: true }));
235
236describe("MergebotQueue", () => {
237 it.effect.each([
238 null,
239 "not JSON",
240 JSON.stringify({ ...configuration, origin: "http://mergebot.example" }),
241 JSON.stringify({ ...configuration, origin: "https://user:password@mergebot.example" }),
242 JSON.stringify({ ...configuration, repositories: ["work/app/extra"] }),
243 ])("disables malformed private configuration: %s", (text) => {
244 const h = harness(text);
245 return Effect.gen(function* () {
246 const service = yield* MergebotQueue.MergebotQueue;
247 expect(service.configuration).toBeNull();
248 expect(yield* service.read({ repository })).toBeNull();
249 expect(h.requests).toHaveLength(0);
250 }).pipe(Effect.provide(h.layer));
251 });
252
253 it.effect(
254 "shares one authenticated stream, derives queue position and never exposes the credential",
255 () => {
256 const h = harness();
257 return Effect.gen(function* () {
258 const service = yield* MergebotQueue.MergebotQueue;
259 const [a, b] = yield* Effect.all(
260 [service.read({ repository }), service.read({ repository: "WORK/APP" })],
261 { concurrency: "unbounded" },
262 );
263 expect(a).toEqual(b);
264 expect(a?.pullRequests?.[0]).toMatchObject({
265 number: 123,
266 position: 1,
267 url: `${url}/pull/123`,
268 });
269 expect(h.requests).toEqual([
270 {
271 url: "wss://mergebot.example/api/state/ws",
272 options: { headers: { Authorization: "Bearer sealed-mergebot-session" } },
273 },
274 ]);
275 expect(JSON.stringify(a)).not.toContain("test-github-token");
276 expect(h.sockets[0]?.sent).toHaveLength(1);
277 expect(yield* service.readPullRequest({ repository, number: 123 })).toMatchObject({
278 headSha: "current-head",
279 checks: [check],
280 });
281 expect(h.requests).toHaveLength(1);
282 expect(h.exchanges).toEqual([
283 {
284 url: "https://mergebot.example/api/auth/github/token",
285 authorization: "Bearer test-github-token",
286 },
287 ]);
288 }).pipe(Effect.provide(h.layer));
289 },
290 );
291
292 it.effect("missing credentials and unapproved repositories never open a socket", () => {
293 const h = harness();
294 h.setToken(null);
295 return Effect.gen(function* () {
296 const service = yield* MergebotQueue.MergebotQueue;
297 expect(yield* service.read({ repository: "other/app" })).toBeNull();
298 expect(yield* service.read({ repository })).toBeNull();
299 expect(yield* service.readPullRequest({ repository, number: 123 })).toBeNull();
300 expect(h.requests).toHaveLength(0);
301 }).pipe(Effect.provide(h.layer));
302 });
303
304 it.effect("cached, loading and unverified detail snapshots cannot satisfy a fresh read", () => {
305 const h = harness();
306 h.subscribeWith((socket, channel) => {
307 if (channel.kind === "repo_overview") socket.snapshot(repoSnapshot);
308 else {
309 socket.snapshot({ ...detailSnapshot, observedAt: null });
310 socket.snapshot({ ...detailSnapshot, observedAt: null });
311 socket.snapshot({
312 ...detailSnapshot,
313 payload: {
314 ...detailSnapshot.payload,
315 view: { ...detailSnapshot.payload.view, isLoadingDetails: true },
316 },
317 });
318 socket.snapshot({
319 ...detailSnapshot,
320 payload: {
321 ...detailSnapshot.payload,
322 entities: { users: {}, pullRequests: { [pr.id]: { ...pr, headSha: "verified-head" } } },
323 },
324 });
325 }
326 });
327 return Effect.gen(function* () {
328 const service = yield* MergebotQueue.MergebotQueue;
329 expect(yield* service.readPullRequest({ repository, number: 123 })).toMatchObject({
330 headSha: "verified-head",
331 position: null,
332 });
333 }).pipe(Effect.provide(h.layer));
334 });
335
336 it.effect("live head changes replace checks and reruns preserve a new failure identity", () => {
337 const h = harness();
338 return Effect.gen(function* () {
339 const service = yield* MergebotQueue.MergebotQueue;
340 yield* service.readPullRequest({ repository, number: 123 });
341 const changed = yield* nextChange(service);
342 const rerun = { ...check, id: "check:2", completedAt: "2026-10-14T10:05:00Z" };
343 h.sockets[0]!.snapshot({
344 ...detailSnapshot,
345 payload: {
346 ...detailSnapshot.payload,
347 entities: {
348 users: {},
349 pullRequests: { [pr.id]: { ...pr, headSha: "new-head", ciStatus: "failed" } },
350 },
351 view: { ...detailSnapshot.payload.view, checks: [rerun] },
352 },
353 });
354 yield* Fiber.join(changed);
355 expect(yield* service.readPullRequest({ repository, number: 123 })).toMatchObject({
356 headSha: "new-head",
357 ciStatus: "failed",
358 checks: [rerun],
359 });
360 }).pipe(Effect.provide(h.layer));
361 });
362
363 it.effect(
364 "disconnect invalidates every snapshot and reconnect resubscribes before accepting fresh data",
365 () => {
366 const h = harness();
367 return Effect.gen(function* () {
368 const service = yield* MergebotQueue.MergebotQueue;
369 yield* service.read({ repository });
370 yield* service.readPullRequest({ repository, number: 123 });
371 const changed = yield* nextChange(service);
372 h.sockets[0]!.close();
373 yield* Fiber.join(changed);
374 expect(yield* service.read({ repository })).toBeNull();
375 expect(yield* service.readPullRequest({ repository, number: 123 })).toBeNull();
376 yield* TestClock.adjust("1 second");
377 expect(h.requests).toHaveLength(2);
378 expect(h.sockets[1]?.sent).toHaveLength(2);
379 expect(yield* service.readPullRequest({ repository, number: 123 })).toMatchObject({
380 checks: [check],
381 });
382 }).pipe(Effect.provide(h.layer));
383 },
384 );
385
386 it.effect("failed initial reads publish unavailable once across authentication retries", () => {
387 const h = harness();
388 h.setExchangeStatus(503);
389 return Effect.gen(function* () {
390 const service = yield* MergebotQueue.MergebotQueue;
391 let changes = 0;
392 yield* service.streamChanges.pipe(
393 Stream.runForEach(() => Effect.sync(() => changes++)),
394 Effect.forkScoped({ startImmediately: true }),
395 );
396 expect(yield* service.read({ repository })).toBeNull();
397 yield* TestClock.adjust("3 seconds");
398 expect(h.exchanges).toHaveLength(4);
399 expect(changes).toBe(1);
400 h.setExchangeStatus(200);
401 const recovered = yield* nextChange(service);
402 yield* TestClock.adjust("1 second");
403 yield* Fiber.join(recovered);
404 expect(yield* service.read({ repository })).toMatchObject({ status: "ready" });
405 expect(changes).toBe(2);
406 }).pipe(Effect.provide(h.layer));
407 });
408
409 it.effect("an outage invalidates ready channels once and publishes fresh recovery", () => {
410 const h = harness();
411 return Effect.gen(function* () {
412 const service = yield* MergebotQueue.MergebotQueue;
413 yield* service.readPullRequest({ repository, number: 123 });
414 const invalidated = yield* Deferred.make<void>();
415 let changes = 0;
416 yield* service.streamChanges.pipe(
417 Stream.runForEach(() =>
418 Effect.gen(function* () {
419 changes++;
420 if (changes === 2) yield* Deferred.succeed(invalidated, undefined);
421 }),
422 ),
423 Effect.forkScoped({ startImmediately: true }),
424 );
425 h.setExchangeStatus(503);
426 h.sockets[0]!.close();
427 yield* Deferred.await(invalidated);
428 expect(yield* service.read({ repository })).toBeNull();
429 expect(yield* service.readPullRequest({ repository, number: 123 })).toBeNull();
430 yield* TestClock.adjust("3 seconds");
431 expect(h.exchanges).toHaveLength(4);
432 expect(changes).toBe(2);
433 h.setExchangeStatus(200);
434 const recovered = yield* nextChange(service);
435 yield* TestClock.adjust("1 second");
436 yield* Fiber.join(recovered);
437 expect(yield* service.readPullRequest({ repository, number: 123 })).toMatchObject({
438 headSha: "current-head",
439 });
440 expect(changes).toBe(4);
441 }).pipe(Effect.provide(h.layer));
442 });
443
444 it.effect("an account change immediately invalidates the old account's snapshot", () => {
445 const h = harness();
446 return Effect.gen(function* () {
447 const service = yield* MergebotQueue.MergebotQueue;
448 yield* service.readPullRequest({ repository, number: 123 });
449 h.setToken("second-account-token");
450 expect(yield* service.readPullRequest({ repository, number: 123 })).toBeNull();
451 yield* TestClock.adjust("1 second");
452 expect(h.exchanges[1]?.authorization).toBe("Bearer second-account-token");
453 expect(h.requests[1]?.options).toEqual({
454 headers: { Authorization: "Bearer sealed-mergebot-session" },
455 });
456 }).pipe(Effect.provide(h.layer));
457 });
458
459 it.effect("channel errors and malformed frames invalidate previously fresh data", () => {
460 const h = harness();
461 return Effect.gen(function* () {
462 const service = yield* MergebotQueue.MergebotQueue;
463 yield* service.readPullRequest({ repository, number: 123 });
464 let changed = yield* nextChange(service);
465 h.sockets[0]!.snapshot({
466 type: "error",
467 channelKey: detailSnapshot.channelKey,
468 message: "Refresh failed",
469 });
470 yield* Fiber.join(changed);
471 expect(yield* service.readPullRequest({ repository, number: 123 })).toBeNull();
472 changed = yield* nextChange(service);
473 h.sockets[0]!.snapshot({ malformed: true });
474 yield* Fiber.join(changed);
475 expect(yield* service.read({ repository })).toBeNull();
476 }).pipe(Effect.provide(h.layer));
477 });
478 it.effect.each([401, 403, 302, 503])(
479 "authentication status %s never opens a socket or reuses ready data",
480 (status) => {
481 const h = harness();
482 h.setExchangeStatus(status);
483 return Effect.gen(function* () {
484 const service = yield* MergebotQueue.MergebotQueue;
485 expect(yield* service.read({ repository })).toBeNull();
486 expect(h.sockets).toHaveLength(0);
487 expect(h.exchanges).toHaveLength(1);
488 }).pipe(Effect.provide(h.layer));
489 },
490 );
491
492 it.effect("a missing verified snapshot has a bounded initial read", () => {
493 const h = harness();
494 h.subscribeWith((socket, channel) => {
495 if (channel.kind === "repo_overview") socket.snapshot(repoSnapshot);
496 else socket.snapshot({ ...detailSnapshot, observedAt: null });
497 });
498 return Effect.gen(function* () {
499 const service = yield* MergebotQueue.MergebotQueue;
500 const read = yield* service
501 .readPullRequest({ repository, number: 123 })
502 .pipe(Effect.forkScoped({ startImmediately: true }));
503 yield* TestClock.adjust("10 seconds");
504 expect(yield* Fiber.join(read)).toBeNull();
505 }).pipe(Effect.provide(h.layer));
506 });
507
508 it.effect("branch discovery uses observed Mergebot metadata and verified detail only", () => {
509 const h = harness();
510 return Effect.gen(function* () {
511 const service = yield* MergebotQueue.MergebotQueue;
512 expect(yield* service.findPullRequest({ repository, headBranch: "feature" })).toMatchObject({
513 number: 123,
514 headSha: "current-head",
515 });
516 expect(yield* service.findPullRequest({ repository, headBranch: "other" })).toBeNull();
517 expect(h.sockets[0]?.sent).toHaveLength(2);
518 }).pipe(Effect.provide(h.layer));
519 });
520
521 it.effect(
522 "a new overview head invalidates prior detail until its own checks are refreshed",
523 () => {
524 const h = harness();
525 return Effect.gen(function* () {
526 const service = yield* MergebotQueue.MergebotQueue;
527 yield* service.readPullRequest({ repository, number: 123 });
528 const changed = yield* nextChange(service);
529 h.sockets[0]!.snapshot({
530 ...repoSnapshot,
531 observedAt: "2026-10-14T10:03:00Z",
532 payload: {
533 ...repoSnapshot.payload,
534 entities: {
535 ...repoSnapshot.payload.entities,
536 pullRequests: { [pr.id]: { ...pr, headSha: "next-head" } },
537 },
538 },
539 });
540 yield* Fiber.join(changed);
541 const read = yield* service
542 .readPullRequest({ repository, number: 123 })
543 .pipe(Effect.forkScoped({ startImmediately: true }));
544 h.sockets[0]!.snapshot({
545 ...detailSnapshot,
546 observedAt: "2026-10-14T10:03:00Z",
547 payload: {
548 ...detailSnapshot.payload,
549 entities: { users: {}, pullRequests: { [pr.id]: { ...pr, headSha: "next-head" } } },
550 view: { ...detailSnapshot.payload.view, checks: [] },
551 },
552 });
553 expect(yield* Fiber.join(read)).toMatchObject({ headSha: "next-head", checks: [] });
554 }).pipe(Effect.provide(h.layer));
555 },
556 );
557
558 it.effect("an older observation cannot replace a newer head or rerun", () => {
559 const h = harness();
560 return Effect.gen(function* () {
561 const service = yield* MergebotQueue.MergebotQueue;
562 yield* service.readPullRequest({ repository, number: 123 });
563 const changed = yield* nextChange(service);
564 h.sockets[0]!.snapshot({
565 ...detailSnapshot,
566 observedAt: "2026-10-14T10:03:00Z",
567 payload: {
568 ...detailSnapshot.payload,
569 view: { ...detailSnapshot.payload.view, checks: [{ ...check, id: "check:2" }] },
570 },
571 });
572 yield* Fiber.join(changed);
573 h.sockets[0]!.snapshot(detailSnapshot);
574 expect(yield* service.readPullRequest({ repository, number: 123 })).toMatchObject({
575 checks: [{ id: "check:2" }],
576 });
577 }).pipe(Effect.provide(h.layer));
578 });
579 it.effect(
580 "queue subscribers receive initial state and live changes without a connection race",
581 () => {
582 const h = harness();
583 return Effect.gen(function* () {
584 const service = yield* MergebotQueue.MergebotQueue;
585 const initial = yield* Deferred.make<void>();
586 const values = yield* service.watchQueue({ repository }).pipe(
587 Stream.tap(() => Deferred.succeed(initial, undefined)),
588 Stream.take(2),
589 Stream.runCollect,
590 Effect.forkScoped({ startImmediately: true }),
591 );
592 yield* Deferred.await(initial);
593 h.sockets[0]!.snapshot({
594 ...repoSnapshot,
595 payload: {
596 ...repoSnapshot.payload,
597 entities: {
598 ...repoSnapshot.payload.entities,
599 repos: { [repository]: { ...repo, processing: false } },
600 },
601 },
602 });
603 expect((yield* Fiber.join(values)).map((queue) => queue?.processing)).toEqual([
604 true,
605 false,
606 ]);
607 expect(h.requests).toHaveLength(1);
608 }).pipe(Effect.provide(h.layer));
609 },
610 );
611 it.effect("later reads retry a failed channel without a reactive retry loop", () => {
612 const h = harness();
613 return Effect.gen(function* () {
614 const service = yield* MergebotQueue.MergebotQueue;
615 yield* service.readPullRequest({ repository, number: 123 });
616 const changed = yield* nextChange(service);
617 h.sockets[0]!.snapshot({
618 type: "error",
619 channelKey: detailSnapshot.channelKey,
620 message: "Upstream unavailable",
621 });
622 yield* Fiber.join(changed);
623 expect(yield* service.readPullRequest({ repository, number: 123 })).toBeNull();
624 expect(h.sockets[0]?.sent).toHaveLength(2);
625 yield* TestClock.adjust("1 second");
626 expect(yield* service.readPullRequest({ repository, number: 123 })).toMatchObject({
627 headSha: "current-head",
628 });
629 expect(h.sockets[0]?.sent).toHaveLength(3);
630 expect(h.requests).toHaveLength(1);
631 }).pipe(Effect.provide(h.layer));
632 });
633 it.effect(
634 "an account change during an initial read cannot return the previous account's result",
635 () => {
636 const h = harness();
637 h.subscribeWith((socket, channel) => {
638 if (channel.kind === "repo_overview") socket.snapshot(repoSnapshot);
639 });
640 return Effect.gen(function* () {
641 const service = yield* MergebotQueue.MergebotQueue;
642 yield* service.read({ repository });
643 const read = yield* service
644 .readPullRequest({ repository, number: 123 })
645 .pipe(Effect.forkScoped({ startImmediately: true }));
646 h.setToken("second-account-token");
647 h.sockets[0]!.snapshot(detailSnapshot);
648 expect(yield* Fiber.join(read)).toBeNull();
649 expect(h.sockets[0]?.readyState).toBe(3);
650 }).pipe(Effect.provide(h.layer));
651 },
652 );
653 it.effect(
654 "idle detail channels release their source subscription and reacquire on demand",
655 () => {
656 const h = harness();
657 return Effect.gen(function* () {
658 const service = yield* MergebotQueue.MergebotQueue;
659 yield* service.readPullRequest({ repository, number: 123 });
660 yield* TestClock.adjust("5 minutes");
661 expect(h.sockets[0]?.sent.at(-1)).toEqual({
662 type: "unsubscribe",
663 channelKey: detailSnapshot.channelKey,
664 });
665 expect(yield* service.readPullRequest({ repository, number: 123 })).toMatchObject({
666 headSha: "current-head",
667 });
668 expect(h.sockets[0]?.sent.at(-1)).toEqual({ type: "subscribe", channel: detailChannel });
669 expect(h.requests).toHaveLength(1);
670 }).pipe(Effect.provide(h.layer));
671 },
672 );
673
674 it.effect("active detail reads renew the lease without replacing the shared stream", () => {
675 const h = harness();
676 return Effect.gen(function* () {
677 const service = yield* MergebotQueue.MergebotQueue;
678 yield* service.readPullRequest({ repository, number: 123 });
679 yield* TestClock.adjust("4 minutes");
680 yield* service.readPullRequest({ repository, number: 123 });
681 yield* TestClock.adjust("2 minutes");
682 expect(h.sockets[0]?.sent).toHaveLength(2);
683 expect(h.requests).toHaveLength(1);
684 }).pipe(Effect.provide(h.layer));
685 });
686 it.effect(
687 "branch-only reads retry an unavailable overview channel without a UI subscriber",
688 () => {
689 const h = harness();
690 return Effect.gen(function* () {
691 const service = yield* MergebotQueue.MergebotQueue;
692 yield* service.findPullRequest({ repository, headBranch: "feature" });
693 const changed = yield* nextChange(service);
694 h.sockets[0]!.snapshot({
695 type: "error",
696 channelKey: repoSnapshot.channelKey,
697 message: "Upstream unavailable",
698 });
699 yield* Fiber.join(changed);
700 expect(yield* service.findPullRequest({ repository, headBranch: "feature" })).toBeNull();
701 yield* TestClock.adjust("1 second");
702 expect(yield* service.findPullRequest({ repository, headBranch: "feature" })).toMatchObject(
703 { number: 123 },
704 );
705 expect(h.sockets[0]?.sent.at(-1)).toEqual({ type: "subscribe", channel: repoChannel });
706 expect(h.requests).toHaveLength(1);
707 }).pipe(Effect.provide(h.layer));
708 },
709 );
710
711 it.effect("mutations reuse the session and enable auto-merge after opening", () => {
712 const h = harness();
713 h.respondWith((request) =>
714 request.url.endsWith("/pulls") ? Response.json({ number: 456 }) : Response.json({ ok: true }),
715 );
716 return Effect.gen(function* () {
717 const service = yield* MergebotQueue.MergebotQueue;
718 expect(
719 yield* service.openPullRequest({
720 repository: "WORK/APP",
721 head: "feature",
722 title: "Ship change",
723 body: "",
724 draft: false,
725 autoMerge: true,
726 }),
727 ).toEqual({ number: 456, url: `${url}/pull/456` });
728 yield* service.updatePullRequest({ repository, number: 456, draft: true });
729 yield* service.setAutoMerge({ repository, number: 456, enabled: false });
730 yield* service.deployProduction({ repository, sha: "abc123" });
731 const api = "https://mergebot.example/api/repos/work/app";
732 const authorization = "Bearer sealed-mergebot-session";
733 expect(h.mutations).toEqual([
734 {
735 method: "POST",
736 url: `${api}/pulls`,
737 authorization,
738 body: { head: "feature", title: "Ship change", body: "", draft: false },
739 },
740 {
741 method: "POST",
742 url: `${api}/pulls/456/auto-merge`,
743 authorization,
744 body: { enabled: true },
745 },
746 { method: "PATCH", url: `${api}/pulls/456`, authorization, body: { draft: true } },
747 {
748 method: "POST",
749 url: `${api}/pulls/456/auto-merge`,
750 authorization,
751 body: { enabled: false },
752 },
753 { method: "POST", url: `${api}/deploy/production`, authorization, body: { sha: "abc123" } },
754 ]);
755 expect(h.exchanges).toHaveLength(1);
756 expect(h.sockets).toHaveLength(0);
757 }).pipe(Effect.provide(h.layer));
758 });
759
760 it.effect("a rejected session is exchanged once and failures carry Mergebot's message", () => {
761 const h = harness();
762 h.setSession("stale-session");
763 h.respondWith((request) => {
764 if (request.authorization !== "Bearer stale-session")
765 return Response.json({ message: "Reference update failed" }, { status: 422 });
766 h.setSession("fresh-session");
767 return new Response(null, { status: 401 });
768 });
769 return Effect.gen(function* () {
770 const service = yield* MergebotQueue.MergebotQueue;
771 const error = yield* service
772 .deployProduction({ repository, sha: "abc123" })
773 .pipe(Effect.flip);
774 expect(error).toMatchObject({
775 _tag: "MergebotOperationError",
776 operation: "deploy to production",
777 status: 422,
778 detail: "Reference update failed",
779 });
780 expect(h.mutations.map((request) => request.authorization)).toEqual([
781 "Bearer stale-session",
782 "Bearer fresh-session",
783 ]);
784 expect(h.exchanges).toHaveLength(2);
785 expect(
786 yield* service
787 .deployProduction({ repository: "other/app", sha: "abc123" })
788 .pipe(Effect.flip),
789 ).toMatchObject({ status: null, detail: "other/app is not a Mergebot repository" });
790 expect(h.mutations).toHaveLength(2);
791 }).pipe(Effect.provide(h.layer));
792 });
793
794 it.effect("an auto-merge failure does not fail the pull request it follows", () => {
795 const h = harness();
796 h.respondWith((request) =>
797 request.url.endsWith("/pulls")
798 ? Response.json({ number: 7 })
799 : Response.json({ message: "No trigger label" }, { status: 400 }),
800 );
801 return Effect.gen(function* () {
802 const service = yield* MergebotQueue.MergebotQueue;
803 expect(
804 yield* service.openPullRequest({
805 repository,
806 head: "feature",
807 base: "release",
808 title: "Ship change",
809 body: "Body",
810 draft: true,
811 autoMerge: true,
812 }),
813 ).toEqual({ number: 7, url: `${url}/pull/7` });
814 expect(h.mutations[0]?.body).toEqual({
815 head: "feature",
816 base: "release",
817 title: "Ship change",
818 body: "Body",
819 draft: true,
820 });
821 }).pipe(Effect.provide(h.layer));
822 });
823
824 it.effect("status watchers hold the detail channel and release it after they leave", () => {
825 const h = harness();
826 return Effect.gen(function* () {
827 const service = yield* MergebotQueue.MergebotQueue;
828 const loaded = yield* Deferred.make<unknown>();
829 const watcher = yield* service.watchPullRequestStatus({ repository, number: 123 }).pipe(
830 Stream.filter((status) => status?.checks != null),
831 Stream.runForEach((status) => Deferred.succeed(loaded, status)),
832 Effect.forkScoped({ startImmediately: true }),
833 );
834 expect(yield* Deferred.await(loaded)).toEqual({
835 number: 123,
836 url: `${url}/pull/123`,
837 title: "Ship change",
838 body: null,
839 headSha: "current-head",
840 headBranch: "feature",
841 lifecycle: "open",
842 draft: false,
843 autoMerge: false,
844 ciStatus: "pending",
845 checks: { completed: 1, total: 1, failed: 1 },
846 hasConflicts: false,
847 queueStatus: "waiting_ci",
848 position: 1,
849 mergeEtaAt: null,
850 deploy: null,
851 });
852 const unsubscribe = { type: "unsubscribe", channelKey: detailSnapshot.channelKey };
853 yield* TestClock.adjust("6 minutes");
854 expect(h.sockets[0]?.sent).not.toContainEqual(unsubscribe);
855 yield* Fiber.interrupt(watcher);
856 yield* TestClock.adjust("5 minutes");
857 expect(h.sockets[0]?.sent.at(-1)).toEqual(unsubscribe);
858 }).pipe(Effect.provide(h.layer));
859 });
860
861 it.effect("deployment watchers map the overview's repository record", () => {
862 const h = harness();
863 const stagingDeployment = {
864 sha: "merged-sha",
865 deployedSha: "merged-sha",
866 mode: "branch_ci",
867 status: "passed",
868 currentStep: null,
869 etaAt: null,
870 checksUrl: "https://github.com/work/app/commit/merged-sha/checks",
871 };
872 const merged = { ...pr, id: `${repository}#100`, number: 100, lifecycle: "merged" };
873 h.subscribeWith((socket, channel) => {
874 if (channel.kind !== "repo_overview") return;
875 socket.snapshot({
876 ...repoSnapshot,
877 payload: {
878 entities: {
879 repos: { [repository]: { ...repo, stagingDeployment } },
880 users: {},
881 pullRequests: { [pr.id]: pr, [merged.id]: merged },
882 mergedCommits: {
883 "merged-sha": {
884 id: "merged-sha",
885 repoId: repository,
886 sha: "merged-sha",
887 pullRequestNumber: 100,
888 authorId: null,
889 authorName: "Octo",
890 message: "Merged change",
891 mergedAt: "2026-10-14T09:00:00Z",
892 url: "https://github.com/work/app/pull/100",
893 labels: [],
894 status: "default",
895 isProductionHead: false,
896 },
897 },
898 },
899 view: { repoId: repository, queuePrIds: [pr.id], mergedCommitIds: ["merged-sha"] },
900 },
901 });
902 });
903 return Effect.gen(function* () {
904 const service = yield* MergebotQueue.MergebotQueue;
905 const deployments = yield* Stream.runHead(service.watchDeployments({ repository }));
906 expect(deployments).toMatchObject({
907 _tag: "Some",
908 value: {
909 repository,
910 staging: { sha: "merged-sha", status: "passed" },
911 production: null,
912 mergedCommits: [{ sha: "merged-sha", pullRequestNumber: 100, authorName: "Octo" }],
913 },
914 });
915 const status = yield* Stream.runHead(
916 service.watchPullRequestStatus({ repository, number: 100 }),
917 );
918 expect(status).toMatchObject({
919 _tag: "Some",
920 value: { lifecycle: "merged", deploy: { state: "staging" } },
921 });
922 expect(yield* Stream.runHead(service.watchDeployments({ repository: "other/app" }))).toEqual(
923 expect.objectContaining({ _tag: "Some", value: null }),
924 );
925 }).pipe(Effect.provide(h.layer));
926 });
927});