1
// Copyright (c) 2017-2024 Cloudflare, Inc.
2
// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3
//     https://opensource.org/licenses/Apache-2.0
4
 
5
#include "eventsource.h"
6
 
7
#include "http.h"
8
#include "messagechannel.h"
9
#include "streams/common.h"
10
 
11
#include <workerd/io/features.h>
12
#include <workerd/jsg/exception.h>
13
#include <workerd/util/mimetype.h>
14
 
15
namespace workerd::api {
16
 
17
namespace {
18
class EventSourceSink final: public WritableStreamSink {
19
 public:
20
  EventSourceSink(EventSource& eventSource): eventSource(eventSource) {}
21
 
22
  kj::Promise<void> write(kj::ArrayPtr<const kj::byte> buffer) override {
23
    // The event stream is a new-line delimited format where each line represents an event.
24
    // We need to scan the buffer for end-of-line characters. When we find one, everything before
25
    // it is pushed into the event queue and we keep scanning. If we do not find an end-of-line
26
    // sequence in the remaining input, we buffer it and wait for the next write to continue
27
    // scanning, or until the stream is ended or aborted.
28
 
29
    if (eventSource == kj::none) {
30
      // Write was received after end() or abort() was called.
31
      // We'll just ignore the write.
32
      return kj::READY_NOW;
33
    }
34
 
35
    auto input = buffer.asChars();
36
 
37
    // The stream may or may not begin with the UTF-8 BOM (%xFEFF). If this is the
38
    // first write, we need to check for it and skip it if it is present.
39
    // The BOM is a 3-byte sequence (0xEF, 0xBB, 0xBF) that encodes the Unicode
40
    // codepoint U+FEFF. We only want to check for this once.
41
    if (!bomChecked) {
42
      bomChecked = true;
43
      if (input.size() >= 3 && input[0] == '\xEF' && input[1] == '\xBB' && input[2] == '\xBF') {
44
        input = input.slice(3);
45
      }
46
    }
47
 
48
    while (input != nullptr) {
49
      KJ_IF_SOME(found, findEndOfLine(input)) {
50
        auto prefix = kept.releaseAsArray();
51
        // Feed the line into the processor.
52
        feed(kj::str(prefix, input.first(found.pos)));
53
        input = found.remaining;
54
        // If we've reached the end of the input, input will == nullptr here.
55
      } else {
56
        // No end-of-line found, buffer the input.
57
        kept.addAll(input.begin(), input.end());
58
        input = nullptr;
59
      }
60
    }
61
 
62
    // Release any buffered events to the EventSource
63
    release();
64
 
65
    return kj::READY_NOW;
66
  }
67
 
68
  kj::Promise<void> write(kj::ArrayPtr<const kj::ArrayPtr<const kj::byte>> pieces) override {
69
    for (auto& piece: pieces) {
70
      co_await write(piece);
71
    }
72
    co_return;
73
  }
74
 
75
  kj::Promise<void> end() override {
76
    // The stream has finished. There's really nothing left to do here. Any partially
77
    // filled data will be dropped on the floor.
78
    clear();
79
    return kj::READY_NOW;
80
  }
81
 
82
  void abort(kj::Exception reason) override {
83
    // There's really nothing to do here.
84
    clear();
85
  }
86
 
87
 private:
88
  kj::Maybe<EventSource&> eventSource;
89
 
90
  // Retained bytes to be processed in the next write.
91
  kj::Vector<char> kept;
92
 
93
  // The collected messages that are pending to be dispatched as events
94
  kj::Vector<EventSource::PendingMessage> pendingMessages;
95
 
96
  // The message that is currently being processed.
97
  kj::Maybe<EventSource::PendingMessage> currentPendingMessage;
98
 
99
  // Set to true once the byte-order-mark has been checked
100
  bool bomChecked = false;
101
 
102
  EventSource::PendingMessage& getPendingMessage() {
103
    KJ_IF_SOME(pending, currentPendingMessage) {
104
      return pending;
105
    }
106
    return currentPendingMessage.emplace();
107
  }
108
 
109
  void feed(kj::String line) {
110
    // Parse line according to the event stream format and dispatch the event.
111
 
112
    // stream        = [ bom ] *event
113
    // event         = *( comment / field ) end-of-line
114
    // comment       = colon *any-char end-of-line
115
    // field         = 1*name-char [ colon [ space ] *any-char ] end-of-line
116
    // end-of-line   = ( cr lf / cr / lf )
117
 
118
    // ; characters
119
    // lf            = %x000A ; U+000A LINE FEED (LF)
120
    // cr            = %x000D ; U+000D CARRIAGE RETURN (CR)
121
    // space         = %x0020 ; U+0020 SPACE
122
    // colon         = %x003A ; U+003A COLON (:)
123
    // bom           = %xFEFF ; U+FEFF BYTE ORDER MARK
124
    // name-char     = %x0000-0009 / %x000B-000C / %x000E-0039 / %x003B-10FFFF
125
    //                 ; a scalar value other than U+000A LINE FEED (LF), U+000D CARRIAGE RETURN
126
    //                   (CR), or U+003A COLON (:)
127
    // any-char      = %x0000-0009 / %x000B-000C / %x000E-10FFFF
128
    //                 ; a scalar value other than U+000A LINE FEED (LF) or U+000D CARRIAGE
129
    //                   RETURN (CR)
130
 
131
    // Note that the BOM (if present) is filtered out in the write() method.
132
 
133
    if (line.size() == 0) {
134
      // Dispatch the current pending message and clear it. If there is no
135
      // pending message, we'll just ignore the line.
136
      KJ_IF_SOME(pending, currentPendingMessage) {
137
        // This message is done and ready to be dispatched. Add it to the
138
        // pendingMessages list. The next time release() is called, it will
139
        // be passed off to the EventSource.
140
        pending.id = kj::str(KJ_ASSERT_NONNULL(eventSource).getLastEventId());
141
        pendingMessages.add(kj::mv(pending));
142
        currentPendingMessage = kj::none;
143
      }
144
    } else if (line[0] == ':') {
145
      // Ignore the line.
146
    } else {
147
      static constexpr auto handle = [](auto& self, kj::ArrayPtr<const char> field,
148
                                         kj::ArrayPtr<const char> value) {
149
        auto& pending = self.getPendingMessage();
150
        auto& ev = KJ_ASSERT_NONNULL(self.eventSource);
151
        // Per the spec, only one space after the colon is optional and trimmed.
152
        // Any other whitespace, or additional spaces aren't accounted for so would
153
        // be part of the value.
154
        if (value.size() > 0 && value[0] == ' ') {
155
          value = value.slice(1);
156
        }
157
        if (field == "data"_kjc) {
158
          pending.data.add(kj::str(value));
159
        } else if (field == "event"_kjc) {
160
          pending.event = kj::str(value);
161
        } else if (field == "id"_kjc) {
162
          ev.setLastEventId(kj::str(value));
163
        } else if (field == "retry"_kjc) {
164
          KJ_IF_SOME(time, kj::str(value).tryParseAs<uint32_t>()) {
165
            KJ_ASSERT_NONNULL(self.eventSource).setReconnectionTime(time);
166
          }
167
          // Ignore the line if it cannot be successfully parsed as a uint32_t
168
        }
169
      };
170
 
171
      KJ_IF_SOME(pos, line.findFirst(':')) {
172
        handle(*this, line.first(pos), line.slice(pos + 1));
173
      } else {
174
        handle(*this, line, ""_kjc);
175
      }
176
    }
177
  }
178
 
179
  void release() {
180
    if (pendingMessages.empty()) return;
181
    auto pending = pendingMessages.releaseAsArray();
182
    // If the event source is gone, just drop the messages on the floor.
183
    KJ_IF_SOME(es, eventSource) {
184
      es.enqueueMessages(kj::mv(pending));
185
    }
186
  }
187
 
188
  void clear() {
189
    eventSource = kj::none;
190
    kept.clear();
191
    pendingMessages.clear();
192
    currentPendingMessage = kj::none;
193
  }
194
 
195
  struct EndOfLine {
196
    size_t pos;
197
    kj::ArrayPtr<const char> remaining;
198
  };
199
  kj::Maybe<EndOfLine> findEndOfLine(kj::ArrayPtr<const char> input) {
200
    // The end-of-line marker is either \n, \r, or \r\n
201
    size_t pos = 0;
202
    while (pos < input.size()) {
203
      if (input[pos] == '\n') {
204
        return EndOfLine{pos, input.slice(pos + 1)};
205
      } else if (input[pos] == '\r') {
206
        if (pos + 1 < input.size() && input[pos + 1] == '\n') {
207
          return EndOfLine{pos, input.slice(pos + 2)};
208
        }
209
        return EndOfLine{pos, input.slice(pos + 1)};
210
      }
211
      pos++;
212
    }
213
    return kj::none;
214
  }
215
};
216
 
217
kj::Promise<void> processBody(IoContext& context, kj::Promise<DeferredProxy<void>> promise) {
218
  try {
219
    co_await context.waitForDeferredProxy(kj::mv(promise));
220
  } catch (...) {
221
    auto ex = kj::getCaughtExceptionAsKj();
222
    // We would see a disconnection exception if the eventstream is closed for
223
    // multiple kinds of reasons. If it's a network error and we know we can
224
    // reconnect, we should try to reconnect.
225
    if (ex.getType() == kj::Exception::Type::DISCONNECTED) {
226
      co_return;
227
    }
228
    // Propagate the exception up.
229
    kj::throwFatalException(kj::mv(ex));
230
  }
231
}
232
}  // namespace
233
 
234
jsg::Ref<EventSource> EventSource::constructor(
235
    jsg::Lock& js, kj::String url, jsg::Optional<EventSourceInit> init) {
236
  JSG_REQUIRE(IoContext::hasCurrent(), DOMNotSupportedError,
237
      "An EventSource can only be created within the context of a worker request.");
238
 
239
  KJ_IF_SOME(i, init) {
240
    KJ_IF_SOME(withCredentials, i.withCredentials) {
241
      JSG_REQUIRE(!withCredentials, DOMNotSupportedError,
242
          "The init.withCredentials option is not supported. It must be false or undefined.");
243
    }
244
  }
245
 
246
  auto eventsource = js.alloc<EventSource>(js,
247
      JSG_REQUIRE_NONNULL(jsg::Url::tryParse(url.asPtr()), DOMSyntaxError,
248
          kj::str("Cannot open an EventSource to '", url, "'. The URL is invalid.")),
249
      kj::mv(init));
250
  eventsource->start(js);
251
  return kj::mv(eventsource);
252
}
253
 
254
jsg::Ref<EventSource> EventSource::from(jsg::Lock& js, jsg::Ref<ReadableStream> readable) {
255
  JSG_REQUIRE(IoContext::hasCurrent(), DOMNotSupportedError,
256
      "An EventSource can only be created within the context of a worker request.");
257
  JSG_REQUIRE(!readable->isLocked(), TypeError, "This ReadableStream is locked.");
258
  JSG_REQUIRE(
259
      !readable->isDisturbed(), TypeError, "This ReadableStream has already been read from.");
260
  auto eventsource = js.alloc<EventSource>(js);
261
  eventsource->run(js, kj::mv(readable), false /* No reconnection attempts */);
262
  return kj::mv(eventsource);
263
}
264
 
265
EventSource::EventSource(jsg::Lock& js, jsg::Url url, kj::Maybe<EventSourceInit> init)
266
    : context(IoContext::current()),
267
      impl({
268
        .url = kj::mv(url),
269
        .options = kj::mv(init).orDefault({}),
270
      }),
271
      abortController(js.alloc<AbortController>(js)),
272
      readyState(State::CONNECTING) {}
273
 
274
EventSource::EventSource(jsg::Lock& js)
275
    : context(IoContext::current()),
276
      abortController(js.alloc<AbortController>(js)),
277
      readyState(State::CONNECTING) {}
278
 
279
void EventSource::notifyError(jsg::Lock& js, const jsg::JsValue& error, bool reconnecting) {
280
  if (readyState == State::CLOSED) return;
281
 
282
  // Abort the connection if it hasn't already been. This will be a non-op if the
283
  // controller has already been aborted.
284
  abortController->abort(js, error);
285
 
286
  if (!reconnecting)
287
    readyState = State::CLOSED;
288
  else
289
    readyState = State::CONNECTING;
290
 
291
  // Dispatch the error event.
292
  dispatchEventImpl(js, js.alloc<ErrorEvent>(js, error));
293
 
294
  // Log the error as an uncaught exception for debugging purposes.
295
  IoContext::current().logUncaughtException(UncaughtExceptionSource::ASYNC_TASK, error);
296
}
297
 
298
void EventSource::notifyOpen(jsg::Lock& js) {
299
  if (readyState == State::CLOSED) return;
300
  readyState = State::OPEN;
301
  dispatchEventImpl(js, js.alloc<OpenEvent>());
302
}
303
 
304
void EventSource::notifyMessages(jsg::Lock& js, kj::Array<PendingMessage> messages) {
305
  if (readyState == State::CLOSED) return;
306
  js.tryCatch([&] {
307
    for (auto& message: messages) {
308
      auto data = kj::str(kj::delimited(kj::mv(message.data), "\n"_kjc));
309
      if (data.size() == 0) continue;
310
      kj::String type = kj::mv(message.event).orDefault([]() { return kj::str("message"); });
311
      dispatchEventImpl(js,
312
          js.alloc<MessageEvent>(js, kj::mv(type), js.str(data), kj::mv(message.id),
313
              kj::none /** source **/, impl.map([](FetchImpl& i) -> jsg::Url& { return i.url; })));
314
    }
315
  }, [&](jsg::Value exception) {
316
    // If we end up with an exception being thrown in one of the event handlers, we will
317
    // stop trying to process the messages and instead just error the EventSource.
318
    notifyError(js, jsg::JsValue(exception.getHandle(js)));
319
  });
320
}
321
 
322
void EventSource::reconnect(jsg::Lock& js) {
323
  KJ_ASSERT(impl != kj::none);
324
  readyState = State::CONNECTING;
325
  abortController = js.alloc<AbortController>(js);
326
  auto signal = abortController->getSignal();
327
  context.awaitIo(js, signal->wrap(js, context.afterLimitTimeout(reconnectionTime)))
328
      .then(js,
329
          JSG_VISITABLE_LAMBDA(
330
              (self = JSG_THIS), (self), (jsg::Lock & js) mutable { self->start(js); }),
331
          JSG_VISITABLE_LAMBDA((self = JSG_THIS), (self), (jsg::Lock& js, jsg::Value exception) {
332
            // In this case, it is most likely the EventSource was closed by the user or
333
            // there was some other failure. We should not continue trying to reconnect.
334
            self->notifyError(js, jsg::JsValue(exception.getHandle(js)));
335
          }));
336
}
337
 
338
void EventSource::start(jsg::Lock& js) {
339
  auto& i = KJ_ASSERT_NONNULL(impl);
340
  if (readyState == State::CLOSED) return;
341
 
342
  auto fetcher = i.options.fetcher.map([](jsg::Ref<Fetcher>& f) { return f.addRef(); });
343
 
344
  static constexpr auto handleError = [](auto& js, auto& self, kj::String message) {
345
    auto ex = js.domException(kj::str("AbortError"), kj::mv(message));
346
    auto handle = KJ_ASSERT_NONNULL(ex.tryGetHandle(js));
347
    self->notifyError(js, jsg::JsValue(handle));
348
    return js.resolvedPromise();
349
  };
350
 
351
  auto onSuccess = JSG_VISITABLE_LAMBDA(
352
      (self = JSG_THIS, fetcher = fetcher.map([](jsg::Ref<Fetcher>& f) -> jsg::Ref<Fetcher> {
353
    return f.addRef();
354
  })),
355
      (self, fetcher), (jsg::Lock& js, jsg::Ref<Response> response) {
356
        if (self->readyState == State::CLOSED) return js.resolvedPromise();
357
        auto& impl = KJ_ASSERT_NONNULL(self->impl);
358
        if (!response->getOk()) {
359
        // Response status code is not 2xx, so we fail.
360
        // No reconnection attempt should be made.
361
        return handleError(
362
            js, self, kj::str("The response status code was ", response->getStatus(), "."));
363
        }
364
 
365
        KJ_IF_SOME(contentType,
366
            response->getHeaders(js)->getCommon(js, capnp::CommonHeaderName::CONTENT_TYPE)) {
367
        bool invalid = false;
368
        KJ_IF_SOME(parsed, MimeType::tryParse(contentType)) {
369
        invalid = parsed != MimeType::EVENT_STREAM;
370
        } else {
371
        invalid = true;
372
        }
373
        if (invalid) {
374
        // No reconnection attempt should be made.
375
        return handleError(js, self, kj::str("The content type '", contentType, "' is invalid."));
376
        }
377
        } else {
378
        // No reconnection attempt should be made.
379
        return handleError(
380
            js, self, kj::str("No content type header was present in the response."));
381
        }
382
 
383
        // If the request was redirected, update the URL to the new location.
384
        if (response->getRedirected()) {
385
        KJ_IF_SOME(newUrl, jsg::Url::tryParse(response->getUrl())) {
386
        impl.url = kj::mv(newUrl);
387
        } else {
388
        }  // Extra else block to squash compiler warning
389
        }
390
 
391
        KJ_IF_SOME(body, response->getBody()) {
392
        // Well, ok! We're ready to start trying to process the stream! We do so by
393
        // pumping the body into an EventSourceSink until the body is closed, canceled,
394
        // or errored.
395
        self->run(js, kj::mv(body), true, response.addRef(), kj::mv(fetcher));
396
        return js.resolvedPromise();
397
        } else {
398
        auto& i = KJ_ASSERT_NONNULL(self->impl);
399
        // If there is no body, there's nothing to do. We'll treat this as if
400
        // the server disconnected. If it only happens once, we'll try to reconnect.
401
        // If it happens again, we'll fail the connection as it is likely indicative
402
        // of a bug in the server or along the path to the server.
403
        if (i.previousNoBody) {
404
        self->notifyError(js, js.error("The server provided no content."));
405
        } else {
406
        i.previousNoBody = true;
407
        self->notifyError(js, js.error("The server provided no content. Will try reconnecting."),
408
            true /* reconnecting */);
409
        self->reconnect(js);
410
        }
411
        return js.resolvedPromise();
412
        }
413
      });
414
 
415
  auto onFailed =
416
      JSG_VISITABLE_LAMBDA((self = JSG_THIS), (self), (jsg::Lock& js, jsg::Value exception) {
417
        self->notifyError(js, jsg::JsValue(exception.getHandle(js)));
418
        return js.resolvedPromise();
419
      });
420
 
421
  auto headers = js.alloc<Headers>();
422
  headers->setCommon(capnp::CommonHeaderName::ACCEPT, MimeType::EVENT_STREAM.essence());
423
  headers->setCommon(capnp::CommonHeaderName::CACHE_CONTROL, kj::str("no-cache"));
424
  if (lastEventId != ""_kjc) {
425
    headers->setUnguarded(js, kj::str("last-event-id"), kj::str(lastEventId));
426
  }
427
 
428
  fetchImpl(js, kj::mv(fetcher), kj::str(i.url),
429
      RequestInitializerDict{
430
        .headers = kj::mv(headers),
431
        .signal = abortController->getSignal(),
432
      })
433
      .then(js, kj::mv(onSuccess), kj::mv(onFailed));
434
}
435
 
436
namespace {
437
template <typename T>
438
kj::Maybe<jsg::Ref<T>> addRef(kj::Maybe<jsg::Ref<T>>& ref) {
439
  return ref.map([](jsg::Ref<T>& r) { return r.addRef(); });
440
}
441
}  // namespace
442
 
443
void EventSource::run(jsg::Lock& js,
444
    jsg::Ref<ReadableStream> readable,
445
    bool withReconnection,
446
    kj::Maybe<jsg::Ref<Response>> response,
447
    kj::Maybe<jsg::Ref<Fetcher>> fetcher) {
448
  notifyOpen(js);
449
 
450
  KJ_IF_SOME(resp, response) {
451
    JSG_REQUIRE(resp->getType() != "error"_kj, TypeError,
452
        "Error responses are unsupported with EventSource");
453
  }
454
 
455
  auto onSuccess =
456
      JSG_VISITABLE_LAMBDA((self = JSG_THIS, readable = readable.addRef(), withReconnection,
457
                               response = addRef(response), fetcher = addRef(fetcher)),
458
          (self, readable, response, fetcher), (jsg::Lock& js) {
459
            // The pump finished. Did the server disconnect? If so, try reconnecting if we can.
460
            self->notifyError(js, js.error("The server disconnected."), withReconnection);
461
            if (withReconnection) self->reconnect(js);
462
          });
463
 
464
  auto onFailed = JSG_VISITABLE_LAMBDA(
465
      (self = JSG_THIS, response = addRef(response), fetcher = addRef(fetcher)),
466
      (self, response, fetcher), (jsg::Lock& js, jsg::Value exception) {
467
        // If the pump fails, catch the error and convert it into an error event.
468
        // If we got here, it likely isn't just a DISCONNECT event. Let's not
469
        // try to reconnect at this point.
470
        self->notifyError(js, jsg::JsValue(exception.getHandle(js)));
471
      });
472
 
473
  // Well, ok! We're ready to start trying to process the stream! We do so by
474
  // pumping the body into an EventSourceSink until the body is closed, canceled,
475
  // or errored.
476
  context
477
      .awaitIo(
478
          js, processBody(context, readable->pumpTo(js, kj::heap<EventSourceSink>(*this), true)))
479
      .then(js, kj::mv(onSuccess), kj::mv(onFailed));
480
}
481
 
482
void EventSource::close(jsg::Lock& js) {
483
  if (closeCalled) return;
484
  closeCalled = true;
485
  abortController->abort(js, kj::none);
486
  readyState = State::CLOSED;
487
}
488
 
489
void EventSource::enqueueMessages(kj::Array<PendingMessage> messages) {
490
  context.addTask(context.run([this, messages = kj::mv(messages)](
491
                                  auto& lock) mutable { notifyMessages(lock, kj::mv(messages)); }));
492
}
493
 
494
void EventSource::setReconnectionTime(uint32_t time) {
495
  // We enforce both a min and max reconnection time. The minimum is 1 second,
496
  // and the maximum is 10 seconds.
497
  reconnectionTime =
498
      kj::max(kj::min(time, MAX_RECONNECTION_TIME), MIN_RECONNECTION_TIME) * kj::MILLISECONDS;
499
}
500
 
501
kj::StringPtr EventSource::getLastEventId() {
502
  return lastEventId;
503
}
504
 
505
void EventSource::setLastEventId(kj::String id) {
506
  lastEventId = kj::mv(id);
507
}
508
 
509
void EventSource::visitForGc(jsg::GcVisitor& visitor) {
510
  KJ_IF_SOME(i, impl) {
511
    visitor.visit(i.options.fetcher);
512
  }
513
  visitor.visit(abortController);
514
}
515
 
516
void EventSource::visitForMemoryInfo(jsg::MemoryTracker& tracker) const {
517
  KJ_IF_SOME(i, impl) {
518
    tracker.trackField("fetcher", i.options.fetcher);
519
    tracker.trackField("url", i.url);
520
  }
521
  tracker.trackField("abortController", abortController);
522
  tracker.trackField("lastEventId", lastEventId);
523
}
524
 
525
}  // namespace workerd::api