Skip to content
Merged
1 change: 0 additions & 1 deletion apps/server/src/device/LocalDeviceHost.ts
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@ import * as FileSystem from "effect/FileSystem";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Path from "effect/Path";
import * as Option from "effect/Option";
import * as Ref from "effect/Ref";
import * as Schema from "effect/Schema";
import * as Scope from "effect/Scope";
Expand Down
68 changes: 68 additions & 0 deletions apps/server/src/pullRequest/PullRequestService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,11 @@ import * as NodeServices from "@effect/platform-node/NodeServices";
import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
import * as KeyValueStore from "effect/unstable/persistence/KeyValueStore";
import { assert, it } from "@effect/vitest";
import * as Cause from "effect/Cause";
import * as Clock from "effect/Clock";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
Expand Down Expand Up @@ -3348,6 +3350,72 @@ it.effect("answers a repeated listing from cache, and concurrent readers share o
}),
);

it.effect.each(["list", "listStats"] as const)(
"retries a replacement %s read that joins a canceled lookup",
(operation) =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>();
const stopping = yield* Deferred.make<void>();
const finishShutdown = yield* Deferred.make<void>();
let hostCalls = 0;
const lookup = Effect.gen(function* () {
if (++hostCalls !== 1) return;
yield* Deferred.succeed(started, undefined);
return yield* Effect.never.pipe(
Effect.onInterrupt(() =>
Deferred.succeed(stopping, undefined).pipe(
Effect.andThen(Deferred.await(finishShutdown)),
),
),
);
});
const service = yield* makeService({
projects: [
project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }),
],
providers: [
fakeProvider("github", {
listChangeRequests: () =>
lookup.pipe(
Effect.as({
items: [changeRequest(1, "2026-07-02T00:00:00Z")],
truncated: false,
continues: false,
}),
),
listChangeRequestStats: () =>
lookup.pipe(
Effect.as([{ repository: "acme/web", number: 1, additions: 3, deletions: 1 }]),
),
}),
],
});
const read =
operation === "list"
? service
.list({ state: "open" })
.pipe(Effect.map((result) => result.entries.map((entry) => entry.number)))
: service
.listStats({
refs: [{ projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }],
})
.pipe(Effect.map((result) => result.stats.map((stat) => stat.number)));
const original = yield* Effect.forkScoped(read);
yield* Deferred.await(started);
const cancellation = yield* Effect.forkScoped(Fiber.interrupt(original));
yield* Deferred.await(stopping);
// Join the old entry while its lookup is still finishing cancellation.
const replacement = yield* Effect.forkScoped(read, { startImmediately: true });
yield* Deferred.succeed(finishShutdown, undefined);
yield* Fiber.join(cancellation);
const canceled = yield* Fiber.await(original);
assert.isTrue(Exit.isFailure(canceled));
if (Exit.isFailure(canceled)) assert.isTrue(Cause.hasInterruptsOnly(canceled.cause));
assert.deepStrictEqual(yield* Fiber.join(replacement), [1]);
assert.strictEqual(hostCalls, 2);
}),
);

it.effect("shares one cold viewer lookup across distinct concurrent lists", () =>
Effect.gen(function* () {
let viewerCalls = 0;
Expand Down
14 changes: 12 additions & 2 deletions apps/server/src/pullRequest/PullRequestService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import {
} from "@t3tools/shared/sourceControl";
import { normalizeGitRemoteUrl } from "@t3tools/shared/git";
import * as Cache from "effect/Cache";
import * as Cause from "effect/Cause";
import * as Clock from "effect/Clock";
import * as Context from "effect/Context";
import * as DateTime from "effect/DateTime";
Expand Down Expand Up @@ -2887,7 +2888,12 @@ export const make = Effect.gen(function* () {
? null
: Object.entries(input.cursors).toSorted(([left], [right]) => left.localeCompare(right)),
]);
return Cache.get(listCache, key);
// A replacement reader can join a lookup still finishing its previous reader's cancellation.
return Cache.get(listCache, key).pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause) ? Cache.get(listCache, key) : Effect.failCause(cause),
),
);
};

const checksCache = yield* Cache.makeWith(
Expand Down Expand Up @@ -3134,7 +3140,11 @@ export const make = Effect.gen(function* () {
}
if (missing.size === 0) return { stats: held };
const key = statsBatchKey(missing.values());
const { result, at } = yield* Cache.get(listStatsCache, key);
const { result, at } = yield* Cache.get(listStatsCache, key).pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause) ? Cache.get(listStatsCache, key) : Effect.failCause(cause),
),
);
for (const [key, ref] of missing) {
const stat = result.stats.find(
(stat) =>
Expand Down
31 changes: 1 addition & 30 deletions apps/web/src/components/pullRequest/PullRequestSpeedActions.tsx
Original file line number Diff line number Diff line change
@@ -1,8 +1,4 @@
import type { PullRequestAction } from "@t3tools/contracts";
import { Effect } from "effect";
import { AtomRegistry } from "effect/unstable/reactivity";
import { appAtomRegistry } from "~/rpc/atomRegistry";
import { pullRequestEnvironment, pullRequestStackAtom } from "~/state/pullRequests";
import { useUiStateStore } from "~/uiStateStore";
import { Button } from "../ui/button";
import { Spinner } from "../ui/spinner";
Expand Down Expand Up @@ -44,32 +40,7 @@ export function PullRequestSpeedActions({
environmentId: entry.environmentId,
reference,
onSuccess: (action) => onActed({ entry, action }),
resolveMergeMethod: async () => {
const target = { environmentId: entry.environmentId, input: reference };
const detailAtom = pullRequestEnvironment.detail({
...target,
input: { ...reference, allowStale: false },
});
appAtomRegistry.refresh(detailAtom);
const detail = await Effect.runPromise(
AtomRegistry.getResult(appAtomRegistry, detailAtom, { suspendOnWaiting: true }),
);
if (
detail.state !== "open" ||
detail.isDraft ||
!detail.capabilities.actions.includes("merge") ||
!detail.viewerPermissions.actions.includes("merge")
) {
throw new Error("This pull request cannot be merged.");
}
if (detail.capabilities.stackActions) {
const stackAtom = pullRequestStackAtom(target);
appAtomRegistry.refresh(stackAtom);
const stack = await Effect.runPromise(
AtomRegistry.getResult(appAtomRegistry, stackAtom, { suspendOnWaiting: true }),
);
if (stack !== null) throw new Error("Open this pull request to merge its stack.");
}
resolveMergeMethod: (detail) => {
const allowed = detail.capabilities.mergeMethods.filter(
(method) => detail.mergeCapabilities[method],
);
Expand Down
25 changes: 25 additions & 0 deletions apps/web/src/components/pullRequest/pullRequestList.logic.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1650,6 +1650,31 @@ describe("pull request list overrides", () => {
expect(applyPullRequestOverrides(rows, new Map(), key, "open")).toBe(rows);
});

it.each([
["open", [1]],
["closed", [2]],
["merged", [3]],
["all", [1, 2, 3]],
] as const)("filters host rows by %s after pending actions settle", (state, expected) => {
const rows = [entry(1, "open"), entry(2, "closed"), entry(3, "merged")];
expect(applyPullRequestOverrides(rows, new Map(), key, state).map((row) => row.number)).toEqual(
expected,
);
});

it("keeps a confirmed close hidden when its override is cleared", () => {
const at = 1_000_000;
const rows = [entry(1, "closed"), entry(2, "open"), entry(3, "merged")];
const overrides = new Map([
["#1", { state: "closed" as const, updatedAt: "2026-07-03T00:00:00Z", token: 1, at }],
]);
const settled = settlePullRequestOverrides(overrides, rows, key, at + 5_000);
expect(settled.size).toBe(0);
expect(applyPullRequestOverrides(rows, settled, key, "open").map((row) => row.number)).toEqual([
2,
]);
});

it("hands back the held object for a row a refresh did not change", () => {
const previous = [entry(1, "open"), entry(2, "open")];
const next = [{ ...entry(1, "open") }, { ...entry(2, "open"), state: "merged" as const }];
Expand Down
12 changes: 4 additions & 8 deletions apps/web/src/components/pullRequest/pullRequestList.logic.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1189,18 +1189,14 @@ export function applyPullRequestOverrides<Entry extends PullRequestListEntry>(
keyOf: (entry: Entry) => string,
state: PullRequestListState,
): ReadonlyArray<Entry> {
if (overrides.size === 0) return entries;
if (overrides.size === 0 && state === "all") return entries;
const out: Entry[] = [];
for (const entry of entries) {
const override = overrides.get(keyOf(entry));
if (override === undefined) {
out.push(entry);
continue;
}
if (state !== "all" && override.state !== state) continue;
out.push({ ...entry, ...override });
if (state !== "all" && (override?.state ?? entry.state) !== state) continue;
out.push(override === undefined ? entry : { ...entry, ...override });
}
return out;
return overrides.size === 0 && out.length === entries.length ? entries : out;
}

/**
Expand Down
9 changes: 6 additions & 3 deletions apps/web/src/components/pullRequest/usePullRequestActions.ts
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ export function usePullRequestActionRunner({
reference: PullRequestRef | null;
onSuccess?: (action: PullRequestAction) => void;
/** Small surfaces resolve repository settings on the click, not for every visible row. */
resolveMergeMethod?: () => Promise<PullRequestMergeMethod>;
resolveMergeMethod?: (detail: PullRequestDetail) => PullRequestMergeMethod;
}) {
const runAction = useAtomCommand(pullRequestEnvironment.runAction, { reportFailure: false });
const [actionPending, setActionPending] = useState(false);
Expand All @@ -148,10 +148,13 @@ export function usePullRequestActionRunner({
pendingRef.current = true;
setActionPending(true);
try {
const mergeMethod = method ?? (action === "merge" ? await resolveMergeMethod?.() : undefined);
const result = await runAction({
environmentId,
input: { ...reference, action, ...(mergeMethod ? { mergeMethod } : {}) },
input: {
...reference,
action,
...(method ? { mergeMethod: method } : resolveMergeMethod ? { resolveMergeMethod } : {}),
},
});
if (result._tag === "Failure") throw squashAtomCommandFailure(result);
toastManager.add({ type: "success", title: ACTION_SUCCESS_LABELS[action] });
Expand Down
29 changes: 20 additions & 9 deletions apps/web/src/routes/_chat.pull-requests.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -793,6 +793,10 @@ function PullRequestsRouteView() {
],
);
const baselineQuery = usePullRequestList(baselineTargets);
const baselineEmpty =
baselineQuery.data?.entries.length === 0 &&
!baselineQuery.isPending &&
baselineQuery.error === null;
const facetTargets = useMemo(() => {
if (!filtersOpen) return NO_LIST_TARGETS;
return environmentQueries.map(({ environmentId, projectIds }) => ({
Expand Down Expand Up @@ -961,8 +965,6 @@ function PullRequestsRouteView() {
speedActionRef.current = ({ entry, action }) => {
// Some hosts accept a merge before it completes. Let the next host read declare it merged.
if (action !== "merge") overrideEntry(entry, action);
Comment thread
macroscopeapp[bot] marked this conversation as resolved.
setDetailRefreshToken((token) => token + 1);
refreshListAndStats(undefined, entry.environmentId);
};
const onSpeedAction = useCallback((result: PullRequestSpeedActionResult) => {
speedActionRef.current(result);
Expand Down Expand Up @@ -1005,13 +1007,15 @@ function PullRequestsRouteView() {
// stay — hydrated or previously answered — rather than being dropped for a feed that
// merely settled first.
const partitions =
partitionsWanted && authoredQuery.data !== null && reviewingQuery.data !== null
? { authored: authoredQuery.data.entries, reviewing: reviewingQuery.data.entries }
: current !== null &&
current.environmentKey === environmentKey &&
current.scope === scopeKey
? current.partitions
: undefined;
partitionsWanted && baselineEmpty
? { authored: [], reviewing: [] }
: partitionsWanted && authoredQuery.data !== null && reviewingQuery.data !== null
? { authored: authoredQuery.data.entries, reviewing: reviewingQuery.data.entries }
: current !== null &&
current.environmentKey === environmentKey &&
current.scope === scopeKey
? current.partitions
: undefined;
// A search's answer is the search's, not the workspace's, so only unsearched lists
// persist. Written here where the held partitions are in reach, so a feed settling
// ahead of them cannot overwrite a stored snapshot that already had both groups.
Expand Down Expand Up @@ -1052,6 +1056,7 @@ function PullRequestsRouteView() {
sentQuery,
listQuery.data,
listQuery.isPending,
baselineEmpty,
partitionsWanted,
authoredQuery.data,
reviewingQuery.data,
Expand Down Expand Up @@ -1343,6 +1348,11 @@ function PullRequestsRouteView() {
*/
const groups = useMemo(() => {
if (search.involvement !== "all") return [{ key: "others" as const, label: "", entries }];
// An empty whole-list answer also empties the priority groups. Their old snapshot must
// not restore the last merged rows after the partition reads are no longer mounted.
if (baselineEmpty) {
return groupPullRequestsByInvolvement(entries, viewers);
}
// Until both partitions have answered, the snapshot's stand in — they are yesterday's
// groups, but whole ones, where grouping the feed's first page locally loses every
// authored row older than it. Once the live reads land they take over; with neither,
Expand Down Expand Up @@ -1372,6 +1382,7 @@ function PullRequestsRouteView() {
return partitionPullRequestsWithPriority(entries, authored, reviewing);
}, [
hasLocalFilters,
baselineEmpty,
localFilters,
authoredQuery.data?.entries,
entries,
Expand Down
Loading
Loading