1/**
2 * Encodes and decodes newline-delimited JSON streams in Effect channels.
3 *
4 * NDJSON stores one complete JSON value on each line. This module has helpers
5 * for byte streams, string streams, and schema-checked records, so streaming
6 * code can read or write one JSON record at a time.
7 *
8 * @stability unstable
9 * @since 4.0.0
10 */
11import * as Arr from "../Array.ts"
12import * as Channel from "../Channel.ts"
13import * as ChannelSchema from "../ChannelSchema.ts"
14import * as Data from "../Data.ts"
15import * as Effect from "../Effect.ts"
16import { dual, identity } from "../Function.ts"
17import * as Predicate from "../Predicate.ts"
18import type * as Schema from "../Schema.ts"
19
20const NdjsonErrorTypeId = "~effect/encoding/Ndjson/NdjsonError"
21
22const encoder = new TextEncoder()
23
24/**
25 * Error raised when NDJSON encoding or decoding fails.
26 *
27 * **Details**
28 *
29 * The `kind` field identifies whether the failure happened while packing or
30 * unpacking, and `cause` preserves the original error.
31 *
32 * @stability unstable
33 * @category errors
34 * @since 4.0.0
35 */
36export class NdjsonError extends Data.TaggedError("NdjsonError")<{
37 readonly kind: "Pack" | "Unpack"
38 readonly cause: unknown
39}> {
40 /**
41 * Marks this value as an NDJSON encoding or decoding error for runtime guards.
42 *
43 * @stability unstable
44 * @since 4.0.0
45 */
46 readonly [NdjsonErrorTypeId] = NdjsonErrorTypeId
47
48 /**
49 * Uses the failed NDJSON operation as the public message.
50 *
51 * @stability unstable
52 * @since 4.0.0
53 */
54 override get message() {
55 return this.kind
56 }
57}
58
59/**
60 * Checks whether a value is an `NdjsonError`.
61 *
62 * @stability unstable
63 * @category guards
64 * @since 4.0.0
65 */
66export const isNdjsonError = (u: unknown): u is NdjsonError => Predicate.hasProperty(u, NdjsonErrorTypeId)
67
68/**
69 * Creates a channel that encodes chunks of values as NDJSON strings.
70 *
71 * **Details**
72 *
73 * Each input item is `JSON.stringify`-encoded, separated by newlines, and the
74 * output chunk ends with a trailing newline.
75 *
76 * @stability unstable
77 * @category constructors
78 * @since 4.0.0
79 */
80export const encodeString = <IE = never, Done = unknown>(): Channel.Channel<
81 Arr.NonEmptyReadonlyArray<string>,
82 IE | NdjsonError,
83 Done,
84 Arr.NonEmptyReadonlyArray<unknown>,
85 IE,
86 Done
87> =>
88 Channel.fromTransform((upstream, _scope) =>
89 Effect.succeed(Effect.flatMap(upstream, (input) => {
90 try {
91 return Effect.succeed(Arr.of(
92 input.map((item) => {
93 const output = JSON.stringify(item)
94 if (output === undefined) {
95 throw new TypeError("Value cannot be represented as JSON")
96 }
97 return output
98 }).join("\n") + "\n"
99 ))
100 } catch (cause) {
101 return Effect.fail(new NdjsonError({ kind: "Pack", cause }))
102 }
103 }))
104 )
105
106/**
107 * Creates a channel that encodes chunks of values as UTF-8 NDJSON bytes.
108 *
109 * @stability unstable
110 * @category constructors
111 * @since 4.0.0
112 */
113export const encode = <IE = never, Done = unknown>(): Channel.Channel<
114 Arr.NonEmptyReadonlyArray<Uint8Array>,
115 IE | NdjsonError,
116 Done,
117 Arr.NonEmptyReadonlyArray<unknown>,
118 IE,
119 Done
120> => Channel.map(encodeString(), Arr.map((_) => encoder.encode(_)))
121
122/**
123 * Creates an NDJSON byte encoder channel for values of a schema.
124 *
125 * **Details**
126 *
127 * Values are first encoded with the schema and then written as UTF-8
128 * newline-delimited JSON.
129 *
130 * @stability unstable
131 * @category constructors
132 * @since 4.0.0
133 */
134export const encodeSchema = <S extends Schema.Constraint>(
135 schema: S
136) =>
137<IE = never, Done = unknown>(): Channel.Channel<
138 Arr.NonEmptyReadonlyArray<Uint8Array>,
139 NdjsonError | Schema.SchemaError | IE,
140 Done,
141 Arr.NonEmptyReadonlyArray<S["Type"]>,
142 IE,
143 Done,
144 S["EncodingServices"]
145> => Channel.pipeTo(ChannelSchema.encode(schema)(), encode())
146
147/**
148 * Creates an NDJSON string encoder channel for values of a schema.
149 *
150 * **Details**
151 *
152 * Values are first encoded with the schema and then written as newline-delimited
153 * JSON strings.
154 *
155 * @stability unstable
156 * @category constructors
157 * @since 4.0.0
158 */
159export const encodeSchemaString = <S extends Schema.Constraint>(
160 schema: S
161) =>
162<IE = never, Done = unknown>(): Channel.Channel<
163 Arr.NonEmptyReadonlyArray<string>,
164 NdjsonError | Schema.SchemaError | IE,
165 Done,
166 Arr.NonEmptyReadonlyArray<S["Type"]>,
167 IE,
168 Done,
169 S["EncodingServices"]
170> => Channel.pipeTo(ChannelSchema.encode(schema)(), encodeString())
171
172/**
173 * Creates a channel that parses NDJSON string chunks into values.
174 *
175 * **When to use**
176 *
177 * Use when NDJSON input arrives as string chunks and each complete line should
178 * be parsed into a JSON value.
179 *
180 * **Details**
181 *
182 * Lines may span input chunks.
183 *
184 * **Gotchas**
185 *
186 * Set `ignoreEmptyLines` to skip blank lines before calling `JSON.parse`;
187 * otherwise blank lines are parsed and fail as invalid JSON.
188 *
189 * @stability unstable
190 * @category constructors
191 * @since 4.0.0
192 */
193export const decodeString = <IE = never, Done = unknown>(options?: {
194 readonly ignoreEmptyLines?: boolean | undefined
195}): Channel.Channel<
196 Arr.NonEmptyReadonlyArray<unknown>,
197 IE | NdjsonError,
198 Done,
199 Arr.NonEmptyReadonlyArray<string>,
200 IE,
201 Done
202> => {
203 const lines = Channel.splitLines<IE, Done>().pipe(
204 options?.ignoreEmptyLines === true ?
205 Channel.filterArray((line) => line.length > 0) :
206 identity
207 )
208 return Channel.mapEffect(lines, (chunk) => {
209 try {
210 return Effect.succeed(Arr.map(chunk, (line) => JSON.parse(line)))
211 } catch (cause) {
212 return Effect.fail(new NdjsonError({ kind: "Unpack", cause }))
213 }
214 })
215}
216
217/**
218 * Creates a channel that decodes UTF-8 byte chunks and parses them as NDJSON.
219 *
220 * **Details**
221 *
222 * Lines may span input chunks, and `ignoreEmptyLines` controls whether blank
223 * lines are skipped before JSON parsing.
224 *
225 * @stability unstable
226 * @category constructors
227 * @since 4.0.0
228 */
229export const decode = <IE = never, Done = unknown>(options?: {
230 readonly ignoreEmptyLines?: boolean | undefined
231}): Channel.Channel<
232 Arr.NonEmptyReadonlyArray<unknown>,
233 IE | NdjsonError,
234 Done,
235 Arr.NonEmptyReadonlyArray<Uint8Array>,
236 IE,
237 Done
238> => {
239 return Channel.pipeTo(Channel.decodeText(), decodeString(options))
240}
241
242/**
243 * Creates an NDJSON byte decoder channel for values of a schema.
244 *
245 * **Details**
246 *
247 * The channel decodes UTF-8 bytes, parses each NDJSON line, and then decodes
248 * each parsed value with the schema.
249 *
250 * @stability unstable
251 * @category constructors
252 * @since 4.0.0
253 */
254export const decodeSchema = <S extends Schema.Constraint>(
255 schema: S
256) =>
257<IE = never, Done = unknown>(options?: {
258 readonly ignoreEmptyLines?: boolean | undefined
259}): Channel.Channel<
260 Arr.NonEmptyReadonlyArray<S["Type"]>,
261 Schema.SchemaError | NdjsonError | IE,
262 Done,
263 Arr.NonEmptyReadonlyArray<Uint8Array>,
264 IE,
265 Done,
266 S["DecodingServices"]
267> => Channel.pipeTo(decode(options), ChannelSchema.decodeUnknown(schema)())
268
269/**
270 * Creates an NDJSON string decoder channel for values of a schema.
271 *
272 * **Details**
273 *
274 * The channel parses each line as JSON and then decodes each parsed value with
275 * the schema.
276 *
277 * @stability unstable
278 * @category constructors
279 * @since 4.0.0
280 */
281export const decodeSchemaString = <S extends Schema.Constraint>(
282 schema: S
283) =>
284<IE = never, Done = unknown>(options?: {
285 readonly ignoreEmptyLines?: boolean | undefined
286}): Channel.Channel<
287 Arr.NonEmptyReadonlyArray<S["Type"]>,
288 Schema.SchemaError | NdjsonError | IE,
289 Done,
290 Arr.NonEmptyReadonlyArray<string>,
291 IE,
292 Done,
293 S["DecodingServices"]
294> => Channel.pipeTo(decodeString(options), ChannelSchema.decodeUnknown(schema)())
295
296/**
297 * Wraps a bidirectional byte channel with NDJSON encoding and decoding.
298 *
299 * **Details**
300 *
301 * Outgoing values are written as UTF-8 NDJSON bytes, and incoming bytes are
302 * parsed as NDJSON values.
303 *
304 * @stability unstable
305 * @category combinators
306 * @since 4.0.0
307 */
308export const duplex: {
309 (options?: {
310 readonly ignoreEmptyLines?: boolean | undefined
311 }): <R, IE, OE, OutDone, InDone>(
312 self: Channel.Channel<
313 Arr.NonEmptyReadonlyArray<Uint8Array>,
314 OE,
315 OutDone,
316 Arr.NonEmptyReadonlyArray<Uint8Array>,
317 IE | NdjsonError,
318 InDone,
319 R
320 >
321 ) => Channel.Channel<
322 Arr.NonEmptyReadonlyArray<unknown>,
323 NdjsonError | OE,
324 OutDone,
325 Arr.NonEmptyReadonlyArray<unknown>,
326 IE,
327 InDone,
328 R
329 >
330 <R, IE, OE, OutDone, InDone>(
331 self: Channel.Channel<
332 Arr.NonEmptyReadonlyArray<Uint8Array>,
333 OE,
334 OutDone,
335 Arr.NonEmptyReadonlyArray<Uint8Array>,
336 IE | NdjsonError,
337 InDone,
338 R
339 >,
340 options?: {
341 readonly ignoreEmptyLines?: boolean | undefined
342 }
343 ): Channel.Channel<
344 Arr.NonEmptyReadonlyArray<unknown>,
345 NdjsonError | OE,
346 OutDone,
347 Arr.NonEmptyReadonlyArray<unknown>,
348 IE,
349 InDone,
350 R
351 >
352} = dual((args) => Channel.isChannel(args[0]), <R, IE, OE, OutDone, InDone>(
353 self: Channel.Channel<
354 Arr.NonEmptyReadonlyArray<Uint8Array>,
355 OE,
356 OutDone,
357 Arr.NonEmptyReadonlyArray<Uint8Array>,
358 IE | NdjsonError,
359 InDone,
360 R
361 >,
362 options?: {
363 readonly ignoreEmptyLines?: boolean | undefined
364 }
365): Channel.Channel<
366 Arr.NonEmptyReadonlyArray<unknown>,
367 NdjsonError | OE,
368 OutDone,
369 Arr.NonEmptyReadonlyArray<unknown>,
370 IE,
371 InDone,
372 R
373> =>
374 Channel.pipeTo(
375 Channel.pipeTo(encode(), self),
376 decode(options)
377 ))
378
379/**
380 * Wraps a bidirectional string channel with NDJSON encoding and decoding.
381 *
382 * **Details**
383 *
384 * Outgoing values are written as NDJSON strings, and incoming strings are parsed
385 * as NDJSON values.
386 *
387 * @stability unstable
388 * @category combinators
389 * @since 4.0.0
390 */
391export const duplexString: {
392 (options?: {
393 readonly ignoreEmptyLines?: boolean | undefined
394 }): <R, IE, OE, OutDone, InDone>(
395 self: Channel.Channel<
396 Arr.NonEmptyReadonlyArray<string>,
397 OE,
398 OutDone,
399 Arr.NonEmptyReadonlyArray<string>,
400 IE | NdjsonError,
401 InDone,
402 R
403 >
404 ) => Channel.Channel<
405 Arr.NonEmptyReadonlyArray<unknown>,
406 NdjsonError | OE,
407 OutDone,
408 Arr.NonEmptyReadonlyArray<unknown>,
409 IE,
410 InDone,
411 R
412 >
413 <R, IE, OE, OutDone, InDone>(
414 self: Channel.Channel<
415 Arr.NonEmptyReadonlyArray<string>,
416 OE,
417 OutDone,
418 Arr.NonEmptyReadonlyArray<string>,
419 IE | NdjsonError,
420 InDone,
421 R
422 >,
423 options?: {
424 readonly ignoreEmptyLines?: boolean | undefined
425 }
426 ): Channel.Channel<
427 Arr.NonEmptyReadonlyArray<unknown>,
428 NdjsonError | OE,
429 OutDone,
430 Arr.NonEmptyReadonlyArray<unknown>,
431 IE,
432 InDone,
433 R
434 >
435} = dual((args) => Channel.isChannel(args[0]), <R, IE, OE, OutDone, InDone>(
436 self: Channel.Channel<
437 Arr.NonEmptyReadonlyArray<string>,
438 OE,
439 OutDone,
440 Arr.NonEmptyReadonlyArray<string>,
441 IE | NdjsonError,
442 InDone,
443 R
444 >,
445 options?: {
446 readonly ignoreEmptyLines?: boolean | undefined
447 }
448): Channel.Channel<
449 Arr.NonEmptyReadonlyArray<unknown>,
450 NdjsonError | OE,
451 OutDone,
452 Arr.NonEmptyReadonlyArray<unknown>,
453 IE,
454 InDone,
455 R
456> =>
457 Channel.pipeTo(
458 Channel.pipeTo(encodeString(), self),
459 decodeString(options)
460 ))
461
462/**
463 * Wraps a bidirectional byte channel with schema-aware NDJSON encoding and
464 * decoding.
465 *
466 * **Details**
467 *
468 * Values sent to the wrapped channel are encoded with `inputSchema`; bytes
469 * received from it are parsed as NDJSON and decoded with `outputSchema`.
470 *
471 * @stability unstable
472 * @category combinators
473 * @since 4.0.0
474 */
475export const duplexSchema: {
476 <In extends Schema.Constraint, Out extends Schema.Constraint>(
477 options: {
478 readonly inputSchema: In
479 readonly outputSchema: Out
480 readonly ignoreEmptyLines?: boolean | undefined
481 }
482 ): <OutErr, OutDone, InErr, InDone, R>(
483 self: Channel.Channel<
484 Arr.NonEmptyReadonlyArray<Uint8Array>,
485 OutErr,
486 OutDone,
487 Arr.NonEmptyReadonlyArray<Uint8Array>,
488 NdjsonError | Schema.SchemaError | InErr,
489 InDone,
490 R
491 >
492 ) => Channel.Channel<
493 Arr.NonEmptyReadonlyArray<Out["Type"]>,
494 NdjsonError | Schema.SchemaError | OutErr,
495 OutDone,
496 Arr.NonEmptyReadonlyArray<In["Type"]>,
497 InErr,
498 InDone,
499 R | In["EncodingServices"] | Out["DecodingServices"]
500 >
501 <Out extends Schema.Constraint, In extends Schema.Constraint, OutErr, OutDone, InErr, InDone, R>(
502 self: Channel.Channel<
503 Arr.NonEmptyReadonlyArray<Uint8Array>,
504 OutErr,
505 OutDone,
506 Arr.NonEmptyReadonlyArray<Uint8Array>,
507 NdjsonError | Schema.SchemaError | InErr,
508 InDone,
509 R
510 >,
511 options: {
512 readonly inputSchema: In
513 readonly outputSchema: Out
514 readonly ignoreEmptyLines?: boolean | undefined
515 }
516 ): Channel.Channel<
517 Arr.NonEmptyReadonlyArray<Out["Type"]>,
518 NdjsonError | Schema.SchemaError | OutErr,
519 OutDone,
520 Arr.NonEmptyReadonlyArray<In["Type"]>,
521 InErr,
522 InDone,
523 R | In["EncodingServices"] | Out["DecodingServices"]
524 >
525} = dual(2, <Out extends Schema.Constraint, In extends Schema.Constraint, OutErr, OutDone, InErr, InDone, R>(
526 self: Channel.Channel<
527 Arr.NonEmptyReadonlyArray<Uint8Array>,
528 OutErr,
529 OutDone,
530 Arr.NonEmptyReadonlyArray<Uint8Array>,
531 NdjsonError | Schema.SchemaError | InErr,
532 InDone,
533 R
534 >,
535 options: {
536 readonly inputSchema: In
537 readonly outputSchema: Out
538 readonly ignoreEmptyLines?: boolean | undefined
539 }
540): Channel.Channel<
541 Arr.NonEmptyReadonlyArray<Out["Type"]>,
542 NdjsonError | Schema.SchemaError | OutErr,
543 OutDone,
544 Arr.NonEmptyReadonlyArray<In["Type"]>,
545 InErr,
546 InDone,
547 R | In["EncodingServices"] | Out["DecodingServices"]
548> => ChannelSchema.duplexUnknown(duplex(self, options), options))
549
550/**
551 * Wraps a bidirectional string channel with schema-aware NDJSON encoding and
552 * decoding.
553 *
554 * **Details**
555 *
556 * Values sent to the wrapped channel are encoded with `inputSchema`; strings
557 * received from it are parsed as NDJSON and decoded with `outputSchema`.
558 *
559 * @stability unstable
560 * @category combinators
561 * @since 4.0.0
562 */
563export const duplexSchemaString: {
564 <In extends Schema.Constraint, Out extends Schema.Constraint>(
565 options: {
566 readonly inputSchema: In
567 readonly outputSchema: Out
568 readonly ignoreEmptyLines?: boolean | undefined
569 }
570 ): <OutErr, OutDone, InErr, InDone, R>(
571 self: Channel.Channel<
572 Arr.NonEmptyReadonlyArray<string>,
573 OutErr,
574 OutDone,
575 Arr.NonEmptyReadonlyArray<string>,
576 NdjsonError | Schema.SchemaError | InErr,
577 InDone,
578 R
579 >
580 ) => Channel.Channel<
581 Arr.NonEmptyReadonlyArray<Out["Type"]>,
582 NdjsonError | Schema.SchemaError | OutErr,
583 OutDone,
584 Arr.NonEmptyReadonlyArray<In["Type"]>,
585 InErr,
586 InDone,
587 R | In["EncodingServices"] | Out["DecodingServices"]
588 >
589 <Out extends Schema.Constraint, In extends Schema.Constraint, OutErr, OutDone, InErr, InDone, R>(
590 self: Channel.Channel<
591 Arr.NonEmptyReadonlyArray<string>,
592 OutErr,
593 OutDone,
594 Arr.NonEmptyReadonlyArray<string>,
595 NdjsonError | Schema.SchemaError | InErr,
596 InDone,
597 R
598 >,
599 options: {
600 readonly inputSchema: In
601 readonly outputSchema: Out
602 readonly ignoreEmptyLines?: boolean | undefined
603 }
604 ): Channel.Channel<
605 Arr.NonEmptyReadonlyArray<Out["Type"]>,
606 NdjsonError | Schema.SchemaError | OutErr,
607 OutDone,
608 Arr.NonEmptyReadonlyArray<In["Type"]>,
609 InErr,
610 InDone,
611 R | In["EncodingServices"] | Out["DecodingServices"]
612 >
613} = dual(2, <Out extends Schema.Constraint, In extends Schema.Constraint, OutErr, OutDone, InErr, InDone, R>(
614 self: Channel.Channel<
615 Arr.NonEmptyReadonlyArray<string>,
616 OutErr,
617 OutDone,
618 Arr.NonEmptyReadonlyArray<string>,
619 NdjsonError | Schema.SchemaError | InErr,
620 InDone,
621 R
622 >,
623 options: {
624 readonly inputSchema: In
625 readonly outputSchema: Out
626 readonly ignoreEmptyLines?: boolean | undefined
627 }
628): Channel.Channel<
629 Arr.NonEmptyReadonlyArray<Out["Type"]>,
630 NdjsonError | Schema.SchemaError | OutErr,
631 OutDone,
632 Arr.NonEmptyReadonlyArray<In["Type"]>,
633 InErr,
634 InDone,
635 R | In["EncodingServices"] | Out["DecodingServices"]
636> => ChannelSchema.duplexUnknown(duplexString(self, options), options))