Open History to see the last commit
1import {
2 applyCommandOutputFrame,
3 EMPTY_COMMAND_OUTPUT,
4} from "@t3tools/client-runtime/state/command-output";
5import {
6 GUEST_COMMAND_OUTPUT_ROUTE,
7 GUEST_ITEM_ROUTE,
8 GUEST_SHARE_PARAM,
9 OrchestrationV2CommandOutputFrame,
10 OrchestrationV2GetTurnItemResult,
11} from "@t3tools/contracts";
12import * as Cause from "effect/Cause";
13import * as Effect from "effect/Effect";
14import * as Option from "effect/Option";
15import * as Queue from "effect/Queue";
16import * as Schema from "effect/Schema";
17import * as Stream from "effect/Stream";
18import { Atom } from "effect/reactivity";
19
20import type { TurnItemSource } from "../chat/timelineRowContext";
21
22/** What an expanded guest row says when it cannot read the item. */
23class GuestItemReadError extends Schema.TaggedError<GuestItemReadError>()("GuestItemReadError", {
24 message: Schema.String,
25}) {}
26
27const decodeDetail = Schema.decodeUnknownSync(Schema.toCodecJson(OrchestrationV2GetTurnItemResult));
28const decodeFrame = Schema.decodeUnknownOption(
29 Schema.fromJsonString(OrchestrationV2CommandOutputFrame),
30);
31
32const itemUrl = (route: string, shareId: string, itemId: string) =>
33 `${route}?${GUEST_SHARE_PARAM}=${encodeURIComponent(shareId)}&item=${encodeURIComponent(itemId)}`;
34
35/** `item` is null when the thread no longer has the item, or the share ended. */
36async function fetchGuestItemDetail(
37 shareId: string,
38 itemId: string,
39): Promise<OrchestrationV2GetTurnItemResult> {
40 const response = await fetch(itemUrl(GUEST_ITEM_ROUTE, shareId, itemId));
41 if (response.status === 403) return { item: null };
42 if (!response.ok) throw new Error(`Guest item request failed with ${response.status}.`);
43 return decodeDetail(await response.json());
44}
45
46// Keys are JSON so a share id or item id can never run into the next part.
47const guestItemDetailAtom = Atom.family((key: string) => {
48 const [shareId, itemId] = JSON.parse(key) as [string, string, string];
49 return Atom.make(
50 Effect.tryPromise({
51 try: () => fetchGuestItemDetail(shareId, itemId),
52 catch: () => new GuestItemReadError({ message: "Could not load this item." }),
53 }),
54 );
55});
56
57/** Ends after the frame of a settled command, so the browser does not reconnect to it. */
58const guestCommandOutputAtom = Atom.family((key: string) => {
59 const [shareId, itemId] = JSON.parse(key) as [string, string];
60 const frames = Stream.callback<OrchestrationV2CommandOutputFrame, GuestItemReadError>((queue) =>
61 Effect.acquireRelease(
62 Effect.sync(() => {
63 const source = new EventSource(itemUrl(GUEST_COMMAND_OUTPUT_ROUTE, shareId, itemId));
64 source.addEventListener("message", (message) => {
65 const frame = decodeFrame(message.data);
66 if (Option.isNone(frame)) return;
67 Queue.offerUnsafe(queue, frame.value);
68 if (!frame.value.running) {
69 source.close();
70 Queue.endUnsafe(queue);
71 }
72 });
73 // A non-200 response closes the source for good: the share ended or the item is gone.
74 source.addEventListener("error", () => {
75 if (source.readyState === EventSource.CLOSED) {
76 Queue.failCauseUnsafe(
77 queue,
78 Cause.fail(new GuestItemReadError({ message: "Output is no longer available." })),
79 );
80 }
81 });
82 return source;
83 }),
84 (source) => Effect.sync(() => source.close()),
85 ),
86 );
87 // `scan` emits its seed first; skipping it keeps the row "loading" until the first frame.
88 return Atom.make(
89 frames.pipe(
90 Stream.scan(() => EMPTY_COMMAND_OUTPUT, applyCommandOutputFrame),
91 Stream.filter((view) => view !== EMPTY_COMMAND_OUTPUT),
92 ),
93 );
94});
95
96/** Expanded rows on the guest page read through the guest routes, with the page's share cookie. */
97export const guestTurnItemSource = (shareId: string): TurnItemSource => ({
98 detail: ({ itemId, revision }) =>
99 guestItemDetailAtom(JSON.stringify([shareId, itemId, revision])),
100 commandOutput: ({ itemId }) => guestCommandOutputAtom(JSON.stringify([shareId, itemId])),
101});