1/**
2 * @title Decoding and encoding streams
3 *
4 * Use `Stream.pipeThroughChannel` with the `Ndjson` and `SchemaBinary` modules to
5 * decode and encode streams of structured data.
6 */
7import { DateTime, Schema, Stream } from "effect"
8import { Ndjson, SchemaBinary } from "effect/encoding"
9
10// SchemaBinary derives a framed binary decoder directly from a schema.
11export const schemaBinaryDecoder = SchemaBinary.decode(Schema.Struct({
12 id: Schema.Int,
13 name: Schema.String
14}))()
15
16// ---------------------------------------------------------------------------
17// Domain
18// ---------------------------------------------------------------------------
19
20// A log entry schema representing structured log events. In practice these
21// would come from a file, HTTP body, or socket connection.
22// `DateTimeUtcFromString` decodes an ISO-8601 string into a `DateTime.Utc`.
23class LogEntry extends Schema.Class<LogEntry>("LogEntry")({
24 timestamp: Schema.DateTimeUtcFromString,
25 level: Schema.Literals(["info", "warn", "error"]),
26 message: Schema.String
27}) {}
28
29// ---------------------------------------------------------------------------
30// Decoding NDJSON strings → objects
31// ---------------------------------------------------------------------------
32
33// Suppose we receive raw NDJSON text from a file or network socket.
34// `Ndjson.decodeString()` is a Channel that splits incoming strings on
35// newlines and `JSON.parse`s each line.
36// Pipe the stream through the channel with `Stream.pipeThroughChannel`.
37export const decodeUntyped = Stream.make(
38 "{\"timestamp\":\"2025-06-01T00:00:00Z\",\"level\":\"info\",\"message\":\"start\"}\n" +
39 "{\"timestamp\":\"2025-06-01T00:00:01Z\",\"level\":\"error\",\"message\":\"oops\"}\n"
40).pipe(
41 Stream.pipeThroughChannel(Ndjson.decodeString()),
42 Stream.runCollect
43)
44
45// When you need schema validation on top of the raw JSON parse, use
46// `Ndjson.decodeSchemaString(Schema)()`. This decodes each line, parses the
47// JSON, and then validates each value against the schema — all in one channel.
48export const decodeTyped = Stream.make(
49 "{\"timestamp\":\"2025-06-01T00:00:00Z\",\"level\":\"info\",\"message\":\"start\"}\n" +
50 "{\"timestamp\":\"2025-06-01T00:00:01Z\",\"level\":\"error\",\"message\":\"oops\"}\n"
51).pipe(
52 Stream.pipeThroughChannel(Ndjson.decodeSchemaString(LogEntry)()),
53 Stream.runCollect
54)
55
56// ---------------------------------------------------------------------------
57// Encoding objects → NDJSON strings
58// ---------------------------------------------------------------------------
59
60// `Ndjson.encodeString()` serialises each value to a JSON line.
61// The resulting stream emits ready-to-write NDJSON strings.
62export const encodeUntyped = Stream.make(
63 { timestamp: "2025-06-01T00:00:00Z", level: "info", message: "start" },
64 { timestamp: "2025-06-01T00:00:01Z", level: "error", message: "oops" }
65).pipe(
66 Stream.pipeThroughChannel(Ndjson.encodeString()),
67 Stream.runCollect
68)
69
70// `Ndjson.encodeSchemaString(Schema)()` encodes each value through the schema
71// first (applying any transformations such as date formatting), then
72// serialises it to an NDJSON line.
73export const encodeTyped = Stream.make(
74 new LogEntry({
75 timestamp: DateTime.makeUnsafe("2025-06-01T00:00:00Z"),
76 level: "info",
77 message: "start"
78 }),
79 new LogEntry({
80 timestamp: DateTime.makeUnsafe("2025-06-01T00:00:01Z"),
81 level: "error",
82 message: "oops"
83 })
84).pipe(
85 Stream.pipeThroughChannel(Ndjson.encodeSchemaString(LogEntry)()),
86 Stream.runCollect
87)
88
89// ---------------------------------------------------------------------------
90// Binary (Uint8Array) variants
91// ---------------------------------------------------------------------------
92
93// When working with binary I/O (e.g. TCP sockets, file descriptors) use the
94// non-string variants. `Ndjson.decode()` expects `Uint8Array` chunks and
95// handles text decoding internally. `Ndjson.encode()` produces `Uint8Array`
96// output.
97const enc = new TextEncoder()
98
99export const decodeBinary = Stream.make(
100 enc.encode("{\"level\":\"info\",\"message\":\"binary\"}\n")
101).pipe(
102 Stream.pipeThroughChannel(Ndjson.decode()),
103 Stream.runCollect
104)
105
106export const encodeBinary = Stream.make(
107 { level: "info", message: "binary" }
108).pipe(
109 Stream.pipeThroughChannel(Ndjson.encode()),
110 Stream.runCollect
111)
112
113// ---------------------------------------------------------------------------
114// Handling empty lines
115// ---------------------------------------------------------------------------
116
117// NDJSON files sometimes contain blank lines (e.g. trailing newlines or
118// pretty-printed output). Pass `{ ignoreEmptyLines: true }` to skip them
119// instead of raising an `NdjsonError`.
120export const decodeIgnoringBlanks = Stream.make(
121 "{\"ok\":true}\n\n{\"ok\":false}\n"
122).pipe(
123 Stream.pipeThroughChannel(Ndjson.decodeString({ ignoreEmptyLines: true })),
124 Stream.runCollect
125)
126
127// ---------------------------------------------------------------------------
128// Error handling
129// ---------------------------------------------------------------------------
130
131// `Ndjson.NdjsonError` is raised when encoding (`kind: "Pack"`) or decoding
132// (`kind: "Unpack"`) fails. You can catch it with `Stream.catchTag` or
133// `Effect.catchTag`.
134export const handleDecodeErrors = Stream.make("not-valid-json\n").pipe(
135 Stream.pipeThroughChannel(Ndjson.decodeString()),
136 Stream.catchTag("NdjsonError", (err) =>
137 // The `kind` field indicates whether the error occurred during
138 // encoding ("Pack") or decoding ("Unpack"), and `cause` contains
139 // the underlying exception.
140 Stream.succeed({ recovered: true, kind: err.kind })),
141 Stream.runCollect
142)
143
144// ---------------------------------------------------------------------------
145// Realistic pipeline: decode → transform → re-encode
146// ---------------------------------------------------------------------------
147
148// A common pattern is to read NDJSON, transform each record, and write it
149// back as NDJSON. This example filters error-level log entries and re-encodes
150// them.
151const ndjsonInput = "{\"timestamp\":\"2025-06-01T00:00:00Z\",\"level\":\"info\",\"message\":\"ok\"}\n" +
152 "{\"timestamp\":\"2025-06-01T00:00:01Z\",\"level\":\"error\",\"message\":\"fail\"}\n" +
153 "{\"timestamp\":\"2025-06-01T00:00:02Z\",\"level\":\"warn\",\"message\":\"slow\"}\n"
154
155export const filterAndReencode = Stream.make(ndjsonInput).pipe(
156 // Decode each line into a validated LogEntry
157 Stream.pipeThroughChannel(Ndjson.decodeSchemaString(LogEntry)()),
158 // Keep only error-level entries
159 Stream.filter((entry) => entry.level === "error"),
160 // Re-encode the filtered entries back to NDJSON strings
161 Stream.pipeThroughChannel(Ndjson.encodeSchemaString(LogEntry)()),
162 Stream.runCollect
163)