char/avril

git clone https://git.t4t.associates/char/avril

Charlotte Somport fetch extension to typescriptf2c35fd2026-10-19

import {
  ArrayPrototypePush as push,
  getter,
  MathMin as min,
  NumberParseInt,
  ObjectCreate,
  ObjectDefineProperty as define,
  ReflectApply as apply,
  ReflectConstruct as construct,
  RegExpPrototypeExec as exec,
  String,
  StringPrototypeSlice as slice,
  TypedArrayPrototypeGetLength as length,
  uncurryThis,
  WeakMap,
  WeakMapPrototypeSet as set,
} from "ext:core/primordials";
import {
  markPromiseHandled,
  promiseThen as then,
  safePromise,
} from "ext:core/promises";
import { enqueueTask } from "ext:core/tasks";
import { dictionary, required, state, usv } from "ext:web/idl.js";
import { userAgent } from "ext:web/navigator.js";
import {
  addListener,
  createException,
  dispatch,
  Event,
  EventTarget,
  preventDefault,
  removeListener,
} from "ext:web/events.js";
import { registerUncloneable } from "ext:web/serialization.js";
import { MessageEvent } from "ext:web/messaging.js";
import { TextDecoder, TextEncoder } from "ext:encoding/main.js";
import { URL } from "ext:url/main.js";
import { clearTimeout, setTimeout } from "ext:timers/timers.js";
import {
  op_fetch_base,
  op_fetch_cancel,
  op_fetch_create,
  op_fetch_read,
  op_fetch_send,
} from "ext:core/ops";
import { extractMime } from "ext:http_types/mime.js";

const CONNECTING = 0,
  OPEN = 1,
  CLOSED = 2;
type ReadyState = typeof CONNECTING | typeof OPEN | typeof CLOSED;
type EventType = "open" | "message" | "error";
type EventHandler<E extends Event> = (this: EventSource, event: E) => unknown;
interface Handler {
  value: Function | null;
  listener: ((event: Event) => void) | null;
}
interface SourceState {
  self: EventSource;
  url: string;
  origin: string;
  withCredentials: boolean;
  state: ReadyState;
  lastEventId: string;
  delay: number;
  resource: number | undefined;
  timer: number | undefined;
  handlers: Record<EventType, Handler>;
  pending: string;
  skipLF: boolean;
  data: string;
  type: string;
  id: string;
}

const sources = new WeakMap<object, SourceState>();
registerUncloneable(sources);
const streaming = { __proto__: null, stream: true };
const digits = /^[0-9]+$/;
const decode = uncurryThis(TextDecoder.prototype.decode),
  encode = uncurryThis(TextEncoder.prototype.encode),
  urlHref = getter(URL.prototype, "href"),
  urlOrigin = getter(URL.prototype, "origin"),
  urlProtocol = getter(URL.prototype, "protocol");
const encoder = new TextEncoder();
const fromCode = String.fromCharCode,
  codeAt = uncurryThis(String.prototype.charCodeAt),
  indexOf = uncurryThis(String.prototype.indexOf);

function syntaxError(message: string): Error {
  return createException({ message, name: "SyntaxError" });
}

// Reconnect requests carry the raw UTF-8 bytes of the last event ID, which
// op_fetch_send re-encodes as one byte per character.
function byteHeaderValue(value: string): string | null {
  const bytes = encode(encoder, value);
  let result = "";
  for (let i = 0; i < length(bytes); i++) {
    const byte = bytes[i];
    // Valid SSE IDs can contain controls that hyper cannot send in headers.
    if ((byte < 32 && byte !== 9) || byte === 127) return null;
    result += fromCode(byte);
  }
  return result;
}

function contentType(list: readonly [name: string, value: string][]): string {
  let value: string | null = null;
  for (let i = 0; i < list.length; i++) {
    if (list[i][0] === "content-type") {
      value = value === null ? list[i][1] : value + "," + list[i][1];
    }
  }
  return extractMime(value)?.essence ?? "";
}

const finish = () => true;

function failure(s: SourceState): void {
  enqueueTask(() => {
    if (s.state === CLOSED) return;
    s.state = CONNECTING;
    dispatch(s.self, new Event("error"));
    // The handler can close the source synchronously.
    if ((s.state as ReadyState) === CLOSED) return;
    s.timer = setTimeout(() => {
      s.timer = undefined;
      if (s.state === CONNECTING) connect(s);
    }, s.delay);
  }, finish);
}

function queueEvent(
  s: SourceState,
  type: string,
  data: string,
  id: string,
): void {
  const origin = s.origin;
  enqueueTask(() => {
    if (s.state === CLOSED) return;
    dispatch(
      s.self,
      new MessageEvent(type, {
        __proto__: null,
        data,
        origin,
        lastEventId: id,
      }),
    );
  }, finish);
}

function feed(s: SourceState, text: string): void {
  if (text === "") return;
  let start = 0;
  for (let i = 0; i < text.length; i++) {
    const code = codeAt(text, i);
    if (s.skipLF) {
      s.skipLF = false;
      // The lone LF after a chunk-split CR belongs to the same line break.
      if (code === 10) {
        start = i + 1;
        continue;
      }
    }
    if (code === 13 || code === 10) {
      line(s, s.pending + slice(text, start, i));
      s.pending = "";
      start = i + 1;
      s.skipLF = code === 13;
    }
  }
  s.pending += slice(text, start);
}

function line(s: SourceState, value: string): void {
  if (value === "") {
    s.lastEventId = s.id;
    if (s.data === "") {
      s.type = "";
      return;
    }
    const data = slice(s.data, 0, s.data.length - 1),
      type = s.type === "" ? "message" : s.type,
      id = s.lastEventId;
    s.data = "";
    s.type = "";
    queueEvent(s, type, data, id);
    return;
  }
  const colon = indexOf(value, ":");
  let field, body;
  if (colon === -1) {
    field = value;
    body = "";
  } else {
    field = slice(value, 0, colon);
    body = slice(value, colon + 1);
    if (codeAt(body, 0) === 32) body = slice(body, 1);
  }
  if (field === "data") s.data = s.data + body + "\n";
  else if (field === "event") s.type = body;
  else if (field === "id") {
    if (indexOf(body, "\u0000") === -1) s.id = body;
  } else if (field === "retry" && exec(digits, body)) {
    s.delay = min(NumberParseInt(body, 10), 2147483647);
  }
}

async function consume(
  s: SourceState,
  resource: number,
  decoder: TextDecoder,
): Promise<void> {
  try {
    while (s.state !== CLOSED) {
      const bytes = await safePromise(op_fetch_read(resource));
      // Closing can race the pending read.
      if (bytes === null || (s.state as ReadyState) === CLOSED) break;
      feed(s, decode(decoder, bytes, streaming));
    }
  } catch {
    // Interrupted streams follow the same reconnection path as EOF.
  } finally {
    decode(decoder);
    op_fetch_cancel(resource);
    s.resource = undefined;
  }
  if (s.state !== CLOSED) failure(s);
}

function connect(s: SourceState): void {
  const resource = op_fetch_create(null);
  s.resource = resource;
  // Preserve the resume cursor until the server replaces or clears it.
  s.pending = "";
  s.skipLF = false;
  s.data = "";
  s.type = "";
  s.id = s.lastEventId;
  const entries: [name: string, value: string][] = [
    ["accept", "text/event-stream"],
    ["cache-control", "no-cache"],
    ["pragma", "no-cache"],
    ["user-agent", userAgent],
  ];
  if (s.lastEventId !== "") {
    const value = byteHeaderValue(s.lastEventId);
    if (value !== null) push(entries, ["last-event-id", value]);
  }
  const pending = op_fetch_send([
    resource,
    s.url,
    "GET",
    entries,
    null,
    false,
    "follow",
  ]);
  then(
    pending,
    (reply) => {
      if (s.state === CLOSED) return;
      const status = reply[0];
      if (status !== 200 || contentType(reply[2]) !== "text/event-stream") {
        op_fetch_cancel(resource);
        s.resource = undefined;
        enqueueTask(() => {
          if (s.state !== CONNECTING) return;
          s.state = CLOSED;
          dispatch(s.self, new Event("error"));
        }, finish);
        return;
      }
      s.origin = urlOrigin(new URL(reply[3]));
      enqueueTask(() => {
        if (s.state !== CONNECTING) {
          op_fetch_cancel(resource);
          return;
        }
        s.state = OPEN;
        dispatch(s.self, new Event("open"));
      }, finish);
      markPromiseHandled(consume(s, resource, new TextDecoder()));
    },
    () => {
      op_fetch_cancel(resource);
      s.resource = undefined;
      if (s.state === CLOSED) return;
      failure(s);
    },
  );
}

const eventTypes = ["open", "message", "error"] as const;

function handler(type: EventType) {
  return {
    get(this: EventSource): Function | null {
      return state(sources, this).handlers[type].value;
    },
    set(this: EventSource, rawValue: unknown): void {
      const handler = state(sources, this).handlers[type];
      const value = typeof rawValue === "function" ? rawValue : null;
      handler.value = value;
      if (value && !handler.listener) {
        handler.listener = (event) => {
          if (apply(handler.value!, this, [event]) === false) {
            preventDefault(event);
          }
        };
        addListener(this, type, handler.listener);
      } else if (!value && handler.listener) {
        removeListener(this, type, handler.listener);
        handler.listener = null;
      }
    },
    configurable: true,
  };
}

interface EventSource {
  readonly CONNECTING: typeof CONNECTING;
  readonly OPEN: typeof OPEN;
  readonly CLOSED: typeof CLOSED;
  onopen: EventHandler<Event> | null;
  onmessage: EventHandler<MessageEvent> | null;
  onerror: EventHandler<Event> | null;
}

class EventSource extends EventTarget {
  declare static readonly CONNECTING: typeof CONNECTING;
  declare static readonly OPEN: typeof OPEN;
  declare static readonly CLOSED: typeof CLOSED;

  // @ts-expect-error Reflect.construct returns the instance without a guest-replaceable super() call.
  constructor(rawUrl: unknown, init: unknown = {}) {
    required(arguments.length, 1);
    const url = usv(rawUrl);
    const options = dictionary(init);
    const withCredentials = !!options.withCredentials;
    let record;
    try {
      record = new URL(url, op_fetch_base() || undefined);
    } catch {
      throw syntaxError("The URL is invalid");
    }
    const self = construct(EventTarget, [], new.target) as EventSource;
    const handlers = ObjectCreate(null) as Record<EventType, Handler>;
    for (let i = 0; i < eventTypes.length; i++) {
      handlers[eventTypes[i]] = { value: null, listener: null };
    }
    const s: SourceState = {
      self,
      url: urlHref(record),
      origin: urlOrigin(record),
      withCredentials,
      state: CONNECTING,
      lastEventId: "",
      delay: 3000,
      resource: undefined,
      timer: undefined,
      handlers,
      pending: "",
      skipLF: false,
      data: "",
      type: "",
      id: "",
    };
    set(sources, self, s);
    const protocol = urlProtocol(record);
    if (protocol === "http:" || protocol === "https:") connect(s);
    else {
      enqueueTask(() => {
        if (s.state === CLOSED) return;
        s.state = CLOSED;
        dispatch(self, new Event("error"));
      }, finish);
    }
    return self;
  }
  get url() {
    return state(sources, this).url;
  }
  get withCredentials() {
    return state(sources, this).withCredentials;
  }
  get readyState() {
    return state(sources, this).state;
  }
  close(): void {
    const s = state(sources, this);
    if (s.state === CLOSED) return;
    s.state = CLOSED;
    if (s.resource !== undefined) op_fetch_cancel(s.resource);
    s.resource = undefined;
    if (s.timer !== undefined) {
      clearTimeout(s.timer);
      s.timer = undefined;
    }
  }
}

for (
  const [name, value] of [
    ["CONNECTING", 0],
    ["OPEN", 1],
    ["CLOSED", 2],
  ] as const
) {
  define(EventSource, name, { value, enumerable: true });
  define(EventSource.prototype, name, { value, enumerable: true });
}
for (const type of eventTypes) {
  define(EventSource.prototype, `on${type}`, handler(type));
}

export { EventSource };