1import * as Effect from "effect/Effect";
2import * as Stream from "effect/Stream";
3import * as NodeBuffer from "node:buffer";
4
5export interface CollectedUint8StreamText {
6 readonly text: string;
7 readonly truncated: boolean;
8 readonly bytes: number;
9 readonly invalidUtf8: boolean;
10}
11
12export const decodeUtf8 = (
13 bytes: Uint8Array,
14): Pick<CollectedUint8StreamText, "text" | "invalidUtf8"> => ({
15 text: Buffer.from(bytes).toString("utf8"),
16 invalidUtf8: !NodeBuffer.isUtf8(bytes),
17});
18
19interface CollectState {
20 chunks: Uint8Array[];
21 readonly bytes: number;
22 readonly truncated: boolean;
23}
24
25export const collectUint8StreamText = <E>(input: {
26 readonly stream: Stream.Stream<Uint8Array, E>;
27 readonly maxBytes?: number | undefined;
28 readonly truncatedMarker?: string | null | undefined;
29}): Effect.Effect<CollectedUint8StreamText, E> => {
30 const maxBytes = input.maxBytes ?? Number.POSITIVE_INFINITY;
31 const truncatedMarker = input.truncatedMarker ?? "";
32
33 return input.stream.pipe(
34 Stream.runFold(
35 (): CollectState => ({
36 chunks: [],
37 bytes: 0,
38 truncated: false,
39 }),
40 (state, chunk): CollectState => {
41 /*
42 * keep draining after truncation so the child process can exit normally.
43 * its a know issue that on windows killing after the output cap can force an expensive taskkill operation and hurt performance
44 */
45 if (state.truncated) {
46 return state;
47 }
48
49 const remainingBytes = maxBytes - state.bytes;
50 if (remainingBytes <= 0) {
51 return {
52 ...state,
53 truncated: true,
54 };
55 }
56
57 const nextChunk =
58 chunk.byteLength > remainingBytes ? chunk.slice(0, remainingBytes) : chunk;
59 state.chunks.push(nextChunk);
60 const bytes = state.bytes + nextChunk.byteLength;
61 const truncated = chunk.byteLength > remainingBytes;
62
63 return {
64 chunks: state.chunks,
65 bytes,
66 truncated,
67 };
68 },
69 ),
70 Effect.map((state): CollectedUint8StreamText => {
71 const decoded = decodeUtf8(Buffer.concat(state.chunks, state.bytes));
72 return {
73 text:
74 state.truncated && truncatedMarker.length > 0
75 ? `${decoded.text}${truncatedMarker}`
76 : decoded.text,
77 bytes: state.bytes,
78 truncated: state.truncated,
79 invalidUtf8: decoded.invalidUtf8,
80 };
81 }),
82 );
83};