1import { type TerminalSummary, WS_METHODS } from "@t3tools/contracts";
2import * as Stream from "effect/Stream";
3import { Atom } from "effect/reactivity";
4
5import {
6 createAtomCommandScheduler,
7 createEnvironmentRpcCommand,
8 createEnvironmentRpcSubscriptionAtomFamily,
9 createEnvironmentSubscriptionAtomFamily,
10} from "./runtime.ts";
11import type { EnvironmentRegistry } from "../connection/registry.ts";
12import { subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
13import {
14 applyTerminalAttachStreamEvent,
15 applyTerminalMetadataStreamEvent,
16 nextTerminalAttachSeedState,
17} from "./terminalSession.ts";
18
19export function createTerminalEnvironmentAtoms<R, E>(
20 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
21) {
22 const lifecycleScheduler = createAtomCommandScheduler();
23 const resizeScheduler = createAtomCommandScheduler();
24 const terminalThreadKey = ({
25 environmentId,
26 input,
27 }: {
28 readonly environmentId: string;
29 readonly input: { readonly threadId: string; readonly terminalId?: string | undefined };
30 }) => JSON.stringify([environmentId, input.threadId]);
31 const terminalSessionKey = ({
32 environmentId,
33 input,
34 }: {
35 readonly environmentId: string;
36 readonly input: { readonly threadId: string; readonly terminalId?: string | undefined };
37 }) => JSON.stringify([environmentId, input.threadId, input.terminalId ?? null]);
38 const lifecycleConcurrency = { mode: "serial" as const, key: terminalThreadKey };
39 return {
40 observe: createEnvironmentSubscriptionAtomFamily(runtime, {
41 label: "environment-data:terminal:observe",
42 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.terminalObserve>) =>
43 Stream.suspend(() =>
44 subscribe(WS_METHODS.terminalObserve, input).pipe(
45 Stream.scan(nextTerminalAttachSeedState, applyTerminalAttachStreamEvent),
46 ),
47 ),
48 }),
49 attach: createEnvironmentSubscriptionAtomFamily(runtime, {
50 label: "environment-data:terminal:attach",
51 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.terminalAttach>) =>
52 Stream.suspend(() =>
53 subscribe(WS_METHODS.terminalAttach, input).pipe(
54 Stream.scan(nextTerminalAttachSeedState, applyTerminalAttachStreamEvent),
55 ),
56 ),
57 }),
58 events: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
59 label: "environment-data:terminal:events",
60 tag: WS_METHODS.subscribeTerminalEvents,
61 }),
62 metadata: createEnvironmentSubscriptionAtomFamily(runtime, {
63 label: "environment-data:terminal:metadata",
64 subscribe: (_input: null) =>
65 subscribe(WS_METHODS.subscribeTerminalMetadata, {}).pipe(
66 Stream.scan((): ReadonlyArray<TerminalSummary> => [], applyTerminalMetadataStreamEvent),
67 Stream.drop(1),
68 ),
69 }),
70 metadataSnapshot: createEnvironmentSubscriptionAtomFamily(runtime, {
71 label: "environment-data:terminal:metadata-snapshot",
72 completeWhen: () => true,
73 subscribe: (_input: null) =>
74 subscribe(WS_METHODS.subscribeTerminalMetadata, {}).pipe(
75 Stream.filter((event) => event.type === "snapshot"),
76 Stream.map((event) => event.terminals),
77 ),
78 }),
79 open: createEnvironmentRpcCommand(runtime, {
80 label: "environment-data:terminal:open",
81 tag: WS_METHODS.terminalOpen,
82 scheduler: lifecycleScheduler,
83 concurrency: lifecycleConcurrency,
84 }),
85 write: createEnvironmentRpcCommand(runtime, {
86 label: "environment-data:terminal:write",
87 tag: WS_METHODS.terminalWrite,
88 }),
89 resize: createEnvironmentRpcCommand(runtime, {
90 label: "environment-data:terminal:resize",
91 tag: WS_METHODS.terminalResize,
92 scheduler: resizeScheduler,
93 concurrency: { mode: "latest", key: terminalSessionKey },
94 }),
95 clear: createEnvironmentRpcCommand(runtime, {
96 label: "environment-data:terminal:clear",
97 tag: WS_METHODS.terminalClear,
98 scheduler: lifecycleScheduler,
99 concurrency: lifecycleConcurrency,
100 }),
101 restart: createEnvironmentRpcCommand(runtime, {
102 label: "environment-data:terminal:restart",
103 tag: WS_METHODS.terminalRestart,
104 scheduler: lifecycleScheduler,
105 concurrency: lifecycleConcurrency,
106 }),
107 close: createEnvironmentRpcCommand(runtime, {
108 label: "environment-data:terminal:close",
109 tag: WS_METHODS.terminalClose,
110 scheduler: lifecycleScheduler,
111 concurrency: lifecycleConcurrency,
112 }),
113 };
114}
115
116export * from "./terminalSession.ts";