1/**
2 * Parses and renders Server-Sent Events text streams.
3 *
4 * Server-Sent Events, or SSE, are the text format used by `EventSource` for
5 * one-way server-to-client updates. This module includes parsers, encoders,
6 * channel helpers, and schema-based helpers for the `id`, `event`, and `data`
7 * fields of each event.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import type { NonEmptyReadonlyArray } from "../Array.ts"
13import * as Arr from "../Array.ts"
14import * as Cause from "../Cause.ts"
15import * as Channel from "../Channel.ts"
16import * as ChannelSchema from "../ChannelSchema.ts"
17import * as Data from "../Data.ts"
18import * as Duration from "../Duration.ts"
19import * as Effect from "../Effect.ts"
20import { hasProperty } from "../Predicate.ts"
21import * as Pull from "../Pull.ts"
22import * as Result from "../Result.ts"
23import * as Schema from "../Schema.ts"
24import type * as SchemaAST from "../SchemaAST.ts"
25import * as SchemaTransformation from "../SchemaTransformation.ts"
26
27const SseErrorTypeId = "~effect/encoding/Sse/SseError"
28
29/**
30 * Error reason raised when pending Server-Sent Events state exceeds the
31 * configured maximum size.
32 *
33 * @stability unstable
34 * @category errors
35 * @since 4.0.0
36 */
37export class EventTooLarge extends Data.TaggedError("EventTooLarge")<{
38 readonly maxEventSize: number
39}> {
40 override get message() {
41 return `Pending SSE event exceeded the maximum size of ${this.maxEventSize}`
42 }
43}
44
45/**
46 * Union of Server-Sent Events decoding error reasons.
47 *
48 * @stability unstable
49 * @category errors
50 * @since 4.0.0
51 */
52export type SseErrorReason = EventTooLarge
53
54/**
55 * Error raised when decoding a Server-Sent Events stream fails.
56 *
57 * @stability unstable
58 * @category errors
59 * @since 4.0.0
60 */
61export class SseError extends Data.TaggedError("SseError")<{
62 readonly reason: SseErrorReason
63}> {
64 /**
65 * Marks this value as an SSE decoding error.
66 *
67 * @stability unstable
68 * @since 4.0.0
69 */
70 readonly [SseErrorTypeId] = SseErrorTypeId
71
72 /**
73 * Delegates the public message to the underlying SSE error reason.
74 *
75 * @stability unstable
76 * @since 4.0.0
77 */
78 override get message() {
79 return this.reason.message
80 }
81}
82
83/**
84 * Options for decoding Server-Sent Events streams.
85 *
86 * @stability unstable
87 * @category decoding
88 * @since 4.0.0
89 */
90export interface DecodeOptions {
91 /**
92 * Maximum number of string code units retained for a pending event. The default is 10 MiB.
93 */
94 readonly maxEventSize?: number | undefined
95}
96
97const defaultMaxEventSize = 10 * 1024 * 1024
98
99/**
100 * Creates a channel that parses Server-Sent Events text chunks into `Event` values.
101 *
102 * **Details**
103 *
104 * SSE `retry` directives are emitted as `Retry` failures so callers can
105 * reconnect with the requested delay.
106 *
107 * @stability unstable
108 * @category decoding
109 * @since 4.0.0
110 */
111export const decode = <IE, Done>(options?: DecodeOptions): Channel.Channel<
112 NonEmptyReadonlyArray<Event>,
113 IE | Retry | SseError,
114 Done,
115 NonEmptyReadonlyArray<string>,
116 IE,
117 Done
118> =>
119 Channel.fromTransform((upstream, _scope) =>
120 Effect.sync(() => {
121 let buffer: Array<Event> = []
122 let retry: Retry | undefined
123 const parser = makeParser((event) => {
124 if (event._tag === "Retry") {
125 retry = event
126 } else {
127 buffer.push(event)
128 }
129 }, options)
130
131 const pump = Effect.flatMap(upstream, (arr) => {
132 for (let i = 0; i < arr.length; i++) {
133 const error = parser.feed(arr[i])
134 if (error !== undefined) {
135 return Effect.fail(error)
136 }
137 }
138 return Effect.void
139 })
140
141 return Effect.suspend(function loop(): Pull.Pull<NonEmptyReadonlyArray<Event>, IE | Retry | SseError, Done> {
142 if (Arr.isArrayNonEmpty(buffer)) {
143 const out = buffer
144 buffer = []
145 return Effect.succeed(out)
146 } else if (retry) {
147 return Effect.fail(retry)
148 }
149 return Effect.flatMap(pump, loop)
150 })
151 })
152 )
153
154/**
155 * A constraint for schemas that can decode SSE events.
156 *
157 * @stability unstable
158 * @category decoding
159 * @since 4.0.0
160 */
161export interface EventCodec extends
162 Schema.ConstraintCodec<
163 any,
164 {
165 readonly id?: string | undefined
166 readonly event?: string | undefined
167 readonly data: string
168 },
169 unknown,
170 unknown
171 >
172{}
173
174/**
175 * Creates an SSE decoder channel that decodes each parsed event with a schema.
176 *
177 * **Details**
178 *
179 * The schema receives `{ event, data, id? }`, with string `data`. Absent IDs are
180 * omitted, even with default options: use `Schema.optional(Schema.String)`, not
181 * `Schema.UndefinedOr(Schema.String)`, to accept events without an ID.
182 *
183 * With `onExcessProperty: "error"`, declare `event` (default: `"message"`) and
184 * `id` if the stream carries IDs, including inherited IDs.
185 *
186 * @stability unstable
187 * @category decoding
188 * @since 4.0.0
189 */
190export const decodeSchema = <
191 S extends EventCodec,
192 IE,
193 Done
194>(
195 schema: S,
196 options?: DecodeOptions,
197 parseOptions?: SchemaAST.ParseOptions
198): Channel.Channel<
199 NonEmptyReadonlyArray<S["Type"]>,
200 IE | Retry | SseError | Schema.SchemaError,
201 Done,
202 NonEmptyReadonlyArray<string>,
203 IE,
204 Done,
205 S["DecodingServices"]
206> =>
207 Channel.pipeTo(
208 decode<IE, Done>(options),
209 ChannelSchema.decode(
210 Event.pipe(
211 Schema.decodeTo(schema, transformEvent)
212 ),
213 parseOptions
214 )()
215 )
216
217/**
218 * Creates an SSE decoder channel that JSON-decodes each event `data` field with a schema.
219 *
220 * **Details**
221 *
222 * The output preserves the SSE `event` name and optional `id` while replacing
223 * `data` with the decoded value.
224 *
225 * @stability unstable
226 * @category decoding
227 * @since 4.0.0
228 */
229export const decodeDataSchema = <Type, DecodingServices, IE, Done>(
230 schema: Schema.ConstraintDecoder<Type, DecodingServices>,
231 options?: DecodeOptions
232): Channel.Channel<
233 NonEmptyReadonlyArray<{
234 readonly event: string
235 readonly id: string | undefined
236 readonly data: Type
237 }>,
238 IE | Retry | SseError | Schema.SchemaError,
239 Done,
240 NonEmptyReadonlyArray<string>,
241 IE,
242 Done,
243 DecodingServices
244> => {
245 const eventSchema = Schema.Struct({
246 ...EventEncoded.fields,
247 data: Schema.fromJsonString(schema)
248 })
249 return Channel.pipeTo(
250 decode<IE, Done>(options),
251 Channel.map(
252 ChannelSchema.decode(eventSchema)(),
253 Arr.map((event) => ({ ...event, id: event.id }))
254 )
255 )
256}
257
258/**
259 * Creates a stateful Server-Sent Events parser.
260 *
261 * **Details**
262 *
263 * Call `feed` with text chunks to parse `Event` and `Retry` values through the
264 * callback, and call `reset` to clear any buffered event state. `feed` returns
265 * an `SseError` if the pending event exceeds `maxEventSize`.
266 *
267 * @stability unstable
268 * @category decoding
269 * @since 4.0.0
270 */
271export function makeParser(onParse: (event: AnyEvent) => void, options?: DecodeOptions): Parser {
272 const maxEventSize = options?.maxEventSize ?? defaultMaxEventSize
273
274 // Processing state
275 let isFirstChunk: boolean
276 let buffer: string
277 let startingPosition: number
278 let startingFieldLength: number
279 let discardTrailingNewline: boolean
280
281 // Event state
282 let lastEventId: string | undefined
283 let eventName: string | undefined
284 let data: string
285
286 reset()
287 return { feed, reset }
288
289 function reset(): void {
290 isFirstChunk = true
291 buffer = ""
292 startingPosition = 0
293 startingFieldLength = -1
294 discardTrailingNewline = false
295
296 lastEventId = undefined
297 eventName = undefined
298 data = ""
299 }
300
301 function feed(chunk: string): SseError | undefined {
302 buffer = buffer ? buffer + chunk : chunk
303
304 // Strip any UTF-8 byte order mark (BOM) at the start of the stream.
305 // Note that we do not strip any non - UTF8 BOM, as eventsource streams are
306 // always decoded as UTF8 as per the specification.
307 if (isFirstChunk && buffer.startsWith(BOM)) {
308 buffer = buffer.slice(BOM.length)
309 }
310
311 isFirstChunk = false
312
313 // Set up chunk-specific processing state
314 const length = buffer.length
315 let position = 0
316
317 // Read the current buffer byte by byte
318 while (position < length) {
319 // EventSource allows for carriage return + line feed, which means we
320 // need to ignore a linefeed character if the previous character was a
321 // carriage return
322 // @todo refactor to reduce nesting, consider checking previous byte?
323 // @todo but consider multiple chunks etc
324 if (discardTrailingNewline) {
325 if (buffer[position] === "\n") {
326 ++position
327 }
328 discardTrailingNewline = false
329 }
330
331 let lineLength = -1
332 let fieldLength = startingFieldLength
333 let character: string
334
335 for (let index = position + startingPosition; lineLength < 0 && index < length; ++index) {
336 character = buffer[index]
337 if (character === ":" && fieldLength < 0) {
338 fieldLength = index - position
339 } else if (character === "\r") {
340 discardTrailingNewline = true
341 lineLength = index - position
342 } else if (character === "\n") {
343 lineLength = index - position
344 }
345 }
346
347 if (lineLength < 0) {
348 startingPosition = length - position
349 startingFieldLength = fieldLength
350 break
351 } else {
352 startingPosition = 0
353 startingFieldLength = -1
354 }
355
356 parseEventStreamLine(buffer, position, fieldLength, lineLength)
357
358 position += lineLength + 1
359 }
360
361 if (position === length) {
362 // If we consumed the entire buffer to read the event, reset the buffer
363 buffer = ""
364 } else if (position > 0) {
365 // If there are bytes left to process, set the buffer to the unprocessed
366 // portion of the buffer only
367 buffer = buffer.slice(position)
368 }
369
370 if (buffer.length + data.length > maxEventSize) {
371 const error = new SseError({ reason: new EventTooLarge({ maxEventSize }) })
372 reset()
373 return error
374 }
375 }
376
377 function parseEventStreamLine(
378 lineBuffer: string,
379 index: number,
380 fieldLength: number,
381 lineLength: number
382 ) {
383 if (lineLength === 0) {
384 // We reached the last line of this event
385 if (data.length > 0) {
386 onParse({
387 _tag: "Event",
388 id: lastEventId,
389 event: eventName || "message",
390 data: data.slice(0, -1) // remove trailing newline
391 })
392 data = ""
393 }
394 eventName = undefined
395 return
396 }
397
398 const noValue = fieldLength < 0
399 const field = lineBuffer.slice(index, index + (noValue ? lineLength : fieldLength))
400 let step = 0
401
402 if (noValue) {
403 step = lineLength
404 } else if (lineBuffer[index + fieldLength + 1] === " ") {
405 step = fieldLength + 2
406 } else {
407 step = fieldLength + 1
408 }
409
410 const position = index + step
411 const valueLength = lineLength - step
412 const value = lineBuffer.slice(position, position + valueLength).toString()
413
414 if (field === "data") {
415 data += value ? `${value}\n` : "\n"
416 } else if (field === "event") {
417 eventName = value
418 } else if (field === "id" && !value.includes("\u0000")) {
419 lastEventId = value
420 } else if (field === "retry" && /^\d+$/.test(value)) {
421 const retry = parseInt(value, 10)
422 onParse(new Retry({ duration: Duration.millis(retry), lastEventId }))
423 }
424 }
425}
426
427const BOM = "\uFEFF"
428
429/**
430 * Stateful Server-Sent Events parser returned by `makeParser`.
431 *
432 * **Details**
433 *
434 * `feed` accepts additional text chunks and returns an `SseError` when the
435 * configured pending event size is exceeded. `reset` clears buffered parser
436 * state.
437 *
438 * @stability unstable
439 * @category decoding
440 * @since 4.0.0
441 */
442export interface Parser {
443 feed(chunk: string): SseError | undefined
444 reset(): void
445}
446
447/**
448 * Creates a channel that encodes `Event` values as Server-Sent Events text.
449 *
450 * **Details**
451 *
452 * If the upstream channel fails with `Retry`, the retry directive is written and
453 * the encoder completes.
454 *
455 * @stability unstable
456 * @category encoding
457 * @since 4.0.0
458 */
459export const encode = <IE, Done>(): Channel.Channel<
460 NonEmptyReadonlyArray<string>,
461 IE,
462 void,
463 NonEmptyReadonlyArray<Event>,
464 IE | Retry,
465 Done
466> =>
467 Channel.fromTransform((upstream, _scope) =>
468 Effect.sync(() => {
469 let done = false
470 const pull = upstream.pipe(
471 Effect.map(Arr.map(encoder.write)),
472 Effect.catchFilter(Retry.filter as any, (retry: any) => {
473 done = true
474 return Effect.succeed(Arr.of(encoder.write(retry)))
475 }),
476 Pull.catchDone(() => Cause.done())
477 ) as Pull.Pull<Arr.NonEmptyReadonlyArray<string>, IE>
478 return Effect.suspend(() => done ? Cause.done() : pull)
479 })
480 )
481
482/**
483 * Creates an SSE encoder channel for values accepted by a schema.
484 *
485 * **Details**
486 *
487 * Values are schema-encoded to the untagged SSE event shape, transformed to
488 * `Event`, and then written as Server-Sent Events text.
489 *
490 * @stability unstable
491 * @category encoding
492 * @since 4.0.0
493 */
494export const encodeSchema = <
495 S extends EventCodec,
496 IE,
497 Done
498>(schema: S): Channel.Channel<
499 NonEmptyReadonlyArray<string>,
500 IE | Schema.SchemaError,
501 void,
502 NonEmptyReadonlyArray<S["Type"]>,
503 IE | Retry,
504 Done,
505 S["EncodingServices"]
506> =>
507 ChannelSchema.encode(Event.pipe(
508 Schema.decodeTo(schema, transformEvent)
509 ))<IE | Retry, Done>().pipe(
510 Channel.pipeTo(encode())
511 )
512
513/**
514 * Encoder capable of rendering an `Event` or `Retry` value as Server-Sent
515 * Events text.
516 *
517 * @stability unstable
518 * @category encoding
519 * @since 4.0.0
520 */
521export interface Encoder {
522 write(event: AnyEvent): string
523}
524
525/**
526 * Tagged model for a Server-Sent Events message containing the event name, optional event ID, and string data payload.
527 *
528 * @stability unstable
529 * @category models
530 * @since 4.0.0
531 */
532export interface Event {
533 readonly _tag: "Event"
534 readonly event: string
535 readonly id: string | undefined
536 readonly data: string
537}
538
539/**
540 * Schema for the untagged Server-Sent Events payload shape containing an optional `id`, `event`, and string `data` fields.
541 *
542 * @stability unstable
543 * @category models
544 * @since 4.0.0
545 */
546export const EventEncoded: Schema.Struct<{
547 readonly id: Schema.optional<Schema.String>
548 readonly event: Schema.String
549 readonly data: Schema.String
550}> = Schema.Struct({
551 id: Schema.optional(Schema.String),
552 event: Schema.String,
553 data: Schema.String
554})
555
556/**
557 * Schema for the tagged Server-Sent Events message model that adds `_tag: "Event"` to the event name, optional event ID, and string data payload.
558 *
559 * @stability unstable
560 * @category models
561 * @since 4.0.0
562 */
563export const Event: Schema.Struct<{
564 readonly _tag: Schema.tag<"Event">
565 readonly id: Schema.UndefinedOr<Schema.String>
566 readonly event: Schema.String
567 readonly data: Schema.String
568}> = Schema.Struct({
569 _tag: Schema.tag("Event"),
570 id: Schema.UndefinedOr(Schema.String),
571 event: Schema.String,
572 data: Schema.String
573})
574
575/**
576 * Schema for transforming untagged SSE event payloads into tagged `Event`
577 * models.
578 *
579 * @stability unstable
580 * @category models
581 * @since 4.0.0
582 */
583export const transformEvent = SchemaTransformation.transform<{
584 readonly id?: string | undefined
585 readonly event?: string | undefined
586 readonly data: string
587}, {
588 readonly _tag: "Event"
589 readonly id: string | undefined
590 readonly event: string
591 readonly data: string
592}>({
593 decode: (event) =>
594 event.id === undefined
595 ? { event: event.event, data: event.data }
596 : { id: event.id, event: event.event, data: event.data },
597 encode: (event) => ({
598 _tag: "Event",
599 id: event.id,
600 event: event.event ?? "message",
601 data: event.data
602 })
603})
604
605/**
606 * Untagged Server-Sent Events payload shape containing the event name, optional event ID, and string data payload.
607 *
608 * @stability unstable
609 * @category models
610 * @since 4.0.0
611 */
612export interface EventEncoded {
613 readonly event: string
614 readonly id?: string | undefined
615 readonly data: string
616}
617
618const RetryTypeId = "~effect/encoding/Sse/Retry" as const
619
620/**
621 * Represents a Server-Sent Events retry directive.
622 *
623 * **Details**
624 *
625 * Decoders surface this value as a failure to request reconnection after
626 * `duration`; encoders serialize an upstream `Retry` failure as a `retry:` line.
627 *
628 * @stability unstable
629 * @category models
630 * @since 4.0.0
631 */
632export class Retry extends Data.TaggedClass("Retry")<{
633 readonly duration: Duration.Duration
634 readonly lastEventId: string | undefined
635}> {
636 /**
637 * Marks this value as an SSE retry directive for runtime guards.
638 *
639 * @stability unstable
640 * @since 4.0.0
641 */
642 readonly [RetryTypeId]: typeof RetryTypeId = RetryTypeId
643 /**
644 * Returns `true` when the value is an SSE retry directive.
645 *
646 * @stability unstable
647 * @since 4.0.0
648 */
649 static is(u: unknown): u is Retry {
650 return hasProperty(u, RetryTypeId)
651 }
652 /**
653 * Separates SSE retry directives from regular event values.
654 *
655 * @stability unstable
656 * @since 4.0.0
657 */
658 static filter<A>(u: A): Result.Result<Retry, Exclude<A, Retry>> {
659 return Retry.is(u) ? Result.succeed(u) : Result.fail(u as any)
660 }
661}
662
663/**
664 * Union of SSE values that can be rendered by an `Encoder`: regular events and
665 * retry directives.
666 *
667 * @stability unstable
668 * @category models
669 * @since 4.0.0
670 */
671export type AnyEvent = Event | Retry
672
673/**
674 * Default Server-Sent Events encoder.
675 *
676 * **Details**
677 *
678 * It renders `Event` values as `id`, `event`, and `data` lines and renders
679 * `Retry` values as `retry:` directives.
680 *
681 * @stability unstable
682 * @category encoding
683 * @since 4.0.0
684 */
685export const encoder: Encoder = {
686 write(event: AnyEvent): string {
687 switch (event._tag) {
688 case "Event": {
689 let data = ""
690 if (event.id !== undefined) {
691 data += `id: ${event.id}\n`
692 }
693 if (event.event !== "message") {
694 data += `event: ${event.event}\n`
695 }
696 data += `data: ${event.data.replace(/\n/g, "\ndata: ")}\n`
697 return data + "\n"
698 }
699 case "Retry": {
700 return `retry: ${Duration.toMillis(event.duration)}\n\n`
701 }
702 }
703 }
704}