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 // Keep draining after truncation; killing a child on Windows can invoke expensive taskkill.
42 if (state.truncated) {
43 return state;
44 }
45
46 const remainingBytes = maxBytes - state.bytes;
47 if (remainingBytes <= 0) {
48 return {
49 ...state,
50 truncated: true,
51 };
52 }
53
54 const nextChunk =
55 chunk.byteLength > remainingBytes ? chunk.slice(0, remainingBytes) : chunk;
56 state.chunks.push(nextChunk);
57 const bytes = state.bytes + nextChunk.byteLength;
58 const truncated = chunk.byteLength > remainingBytes;
59
60 return {
61 chunks: state.chunks,
62 bytes,
63 truncated,
64 };
65 },
66 ),
67 Effect.map((state): CollectedUint8StreamText => {
68 const decoded = decodeUtf8(Buffer.concat(state.chunks, state.bytes));
69 return {
70 text:
71 state.truncated && truncatedMarker.length > 0
72 ? `${decoded.text}${truncatedMarker}`
73 : decoded.text,
74 bytes: state.bytes,
75 truncated: state.truncated,
76 invalidUtf8: decoded.invalidUtf8,
77 };
78 }),
79 );
80};