/*
 * Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT.
 */

// Some environments' type definitions do not include ReadableStream's async
// iterator, even though the runtime supports it or this class polyfills it.
// Declare the method here so Stream can override it and preserve T in for-await.
declare global {
  interface ReadableStream<R = any> {
    [Symbol.asyncIterator](): {
      next(): Promise<IteratorResult<R, undefined>>;
      throw?(e?: unknown): Promise<IteratorResult<R, undefined>>;
      return?(): Promise<IteratorResult<R, undefined>>;
      [Symbol.asyncIterator](): any;
    };
  }
}

export type SseMessage<T> = {
  data?: T | undefined;
  event?: string | null | undefined;
  id?: string | null | undefined;
  retry?: number | null | undefined;
};
export class EventStream<T extends SseMessage<unknown>>
  extends ReadableStream<T>
{
  constructor(
    responseBody: ReadableStream<Uint8Array>,
    parse: (x: SseMessage<string>) => IteratorResult<T, undefined>,
    opts?: { dataRequired?: boolean },
  ) {
    const upstream = responseBody.getReader();
    let buffer: Uint8Array = new Uint8Array(4096);
    let bufferLen = 0;
    let searchStart = 0;
    const state = { eventId: undefined as string | undefined };
    const dataRequired = opts?.dataRequired ?? true;
    super({
      async pull(downstream) {
        try {
          while (true) {
            const match = findBoundary(buffer, bufferLen, searchStart);
            if (!match) {
              // Bytes before the trailing MAX_BOUNDARY_LEN-1 were already
              // scanned with full lookahead and cannot start a boundary even
              // once more data arrives, so the next scan can skip them.
              searchStart = Math.max(0, bufferLen - MAX_BOUNDARY_LEN + 1);
              const chunk = await upstream.read();
              if (chunk.done) return downstream.close();
              if (bufferLen + chunk.value.length > buffer.length) {
                const grown = new Uint8Array(
                  Math.max(buffer.length * 2, bufferLen + chunk.value.length),
                );
                grown.set(buffer.subarray(0, bufferLen));
                buffer = grown;
              }
              buffer.set(chunk.value, bufferLen);
              bufferLen += chunk.value.length;
              continue;
            }
            const message = buffer.slice(0, match.index);
            buffer.copyWithin(0, match.index + match.length, bufferLen);
            bufferLen -= match.index + match.length;
            if (buffer.length > 4096 && bufferLen <= buffer.length >> 2) {
              // Release oversized capacity retained after an unusually large
              // event so long-lived streams do not hold peak memory.
              const shrunk = new Uint8Array(Math.max(4096, bufferLen * 2));
              shrunk.set(buffer.subarray(0, bufferLen));
              buffer = shrunk;
            }
            searchStart = 0;
            const item = parseMessage(message, parse, state, dataRequired);
            if (item && !item.done) return downstream.enqueue(item.value);
            if (item?.done) {
              await upstream.cancel("done");
              return downstream.close();
            }
          }
        } catch (e) {
          downstream.error(e);
          await upstream.cancel(e);
        }
      },
      cancel: reason => upstream.cancel(reason),
    });
  }

  // Use ReadableStream's iterator return type instead of `any` so stream events
  // keep their generated type in for-await loops.
  override [Symbol.asyncIterator](): ReturnType<
    ReadableStream<T>[typeof Symbol.asyncIterator]
  > {
    const fn = (ReadableStream.prototype as any)[Symbol.asyncIterator];
    if (typeof fn === "function") return fn.call(this);
    const reader = this.getReader();
    return {
      next: async () => {
        const r = await reader.read();
        if (r.done) {
          reader.releaseLock();
          return { done: true, value: undefined };
        }
        return { done: false, value: r.value };
      },
      throw: async (e) => {
        await reader.cancel(e);
        reader.releaseLock();
        return { done: true, value: undefined };
      },
      return: async () => {
        await reader.cancel("done");
        reader.releaseLock();
        return { done: true, value: undefined };
      },
      [Symbol.asyncIterator]() {
        return this;
      },
    } as ReturnType<ReadableStream<T>[typeof Symbol.asyncIterator]>;
  }
}

const CR = 13;
const LF = 10;
const BOUNDARIES = [
  [CR, LF, CR, LF], // \r\n\r\n
  [CR, LF, CR], // \r\n\r
  [CR, LF, LF], // \r\n\n
  [CR, CR, LF], // \r\r\n
  [LF, CR, LF], // \n\r\n
  [CR, CR], // \r\r
  [LF, CR], // \n\r
  [LF, LF], // \n\n
];
const MAX_BOUNDARY_LEN = BOUNDARIES.reduce((m, b) => Math.max(m, b.length), 0);

function findBoundary(
  buf: Uint8Array,
  len: number,
  from: number,
): { index: number; length: number } | null {
  for (let i = from; i < len; i++) {
    if (buf[i] !== CR && buf[i] !== LF) continue;
    for (const boundary of BOUNDARIES) {
      if (i + boundary.length > len) continue;
      let match = true;
      for (let j = 0; j < boundary.length; j++) {
        if (buf[i + j] !== boundary[j]) {
          match = false;
          break;
        }
      }
      if (match) return { index: i, length: boundary.length };
    }
  }
  return null;
}
function parseMessage<T extends SseMessage<unknown>>(
  chunk: Uint8Array,
  parse: (x: SseMessage<string>) => IteratorResult<T, undefined>,
  state: { eventId: string | undefined },
  dataRequired: boolean,
) {
  const text = new TextDecoder().decode(chunk);
  const lines = text.split(/\r\n|\r|\n/);
  const dataLines: string[] = [];
  const ret: SseMessage<string> = {};
  let ignore = true;
  for (const line of lines) {
    if (!line || line.startsWith(":")) continue;
    ignore = false;
    const i = line.indexOf(":");
    let field = line;
    let value = "";
    if (i > 0) {
      field = line.slice(0, i);
      value = line[i + 1] === " " ? line.slice(i + 2) : line.slice(i + 1);
    }
    if (field === "data") dataLines.push(value);
    else if (field === "event") ret.event = value;
    else if (field === "id" && !value.includes("\0")) state.eventId = value;
    else if (field === "retry" && /^\d+$/.test(value)) {
      ret.retry = Number(value);
    }
  }
  if (ignore) return;
  ret.id = state.eventId;
  if (dataLines.length) ret.data = dataLines.join("\n");
  else if (dataRequired) return; // skip data-less events when data is required
  return parse(ret);
}