1import {
2 type ScriptProgressEvent,
3 type ScriptProgressStream,
4 WS_METHODS,
5} from "@t3tools/contracts";
6import * as Stream from "effect/Stream";
7import { Atom } from "effect/reactivity";
8
9import type { EnvironmentRegistry } from "../connection/registry.ts";
10import { createEnvironmentRpcSubscriptionAtomFamily } from "./runtime.ts";
11
12export function applyScriptProgressEvent(
13 streams: ReadonlyArray<ScriptProgressStream>,
14 event: ScriptProgressEvent,
15): ReadonlyArray<ScriptProgressStream> {
16 switch (event.type) {
17 case "snapshot":
18 return event.streams;
19 case "upsert": {
20 const index = streams.findIndex((stream) => stream.id === event.stream.id);
21 return index < 0
22 ? [...streams, event.stream]
23 : streams.map((stream) => (stream.id === event.stream.id ? event.stream : stream));
24 }
25 case "progress":
26 return streams.map((stream) =>
27 stream.id === event.id ? { ...stream, summary: event.summary } : stream,
28 );
29 case "remove":
30 return streams.filter((stream) => stream.id !== event.id);
31 }
32}
33
34export function createScriptProgressEnvironmentAtoms<R, E>(
35 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
36) {
37 return {
38 watch: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
39 label: "environment-data:script-progress:watch",
40 tag: WS_METHODS.subscribeScriptProgress,
41 idleTtlMs: 0,
42 transform: (stream) =>
43 Stream.suspend(() => {
44 let streams: ReadonlyArray<ScriptProgressStream> = [];
45 return stream.pipe(
46 Stream.mapArray((events) => {
47 for (const event of events) streams = applyScriptProgressEvent(streams, event);
48 return [{ events, streams }];
49 }),
50 );
51 }),
52 }),
53 };
54}