// Copyright 2018-2026 the Deno authors. MIT license.
// This module is browser compatible.

/**
 * Async-generator adapter that turns an XML chunk source into a stream of
 * application-defined records assembled inside SAX-style event callbacks.
 *
 * @module
 */

import type { ParseStreamOptions, XmlEventCallbacks } from "./types.ts";
import { createXmlPipeline } from "./_pipeline.ts";
import { decodeAsyncIterable } from "./_decode.ts";

/**
 * Parses an async iterable of XML chunks and yields records assembled inside
 * SAX-style event callbacks.
 *
 * Each input chunk is parsed synchronously and the records emitted from that
 * chunk are buffered before any are yielded; the consumer then pulls records
 * one at a time. Yield rate (and downstream backpressure) is per-record while
 * peak memory is bounded by the records produced by a single chunk — for
 * example, a 1 MB chunk containing 1000 items will buffer all 1000 records
 * before the consumer sees the first.
 *
 * If parsing throws (XML syntax error or a user callback that throws), the
 * iteration rejects immediately; records buffered within the failing chunk
 * are discarded. Records yielded by earlier chunks remain visible.
 *
 * For `pipeThrough` composition, wrap the result with
 * {@linkcode ReadableStream.from}.
 *
 * @typeParam T The type of records yielded by the generator. Must be
 * specified explicitly; cannot be inferred from `emit` calls.
 *
 * @param source An async iterable of XML string chunks. For byte sources,
 * use {@linkcode parseXmlRecordsFromBytes}.
 *
 * @param createCallbacks Factory that builds the SAX callbacks; receives an
 * `emit` function to push completed records.
 *
 * @param options Parser options forwarded to the underlying tokenizer/parser.
 *
 * @returns An async generator that yields records as the document is parsed.
 *
 * @example Parse items from an XML feed into typed records
 * ```ts
 * import { parseXmlRecords } from "@std/xml/parse-records";
 * import { assertEquals } from "@std/assert";
 *
 * type Item = { id: string; title: string };
 *
 * const xml =
 *   `<feed><item id="1" title="First"/><item id="2" title="Second"/></feed>`;
 *
 * const items: Item[] = [];
 * for await (
 *   const item of parseXmlRecords<Item>(
 *     ReadableStream.from([xml]),
 *     (emit) => ({
 *       onStartElement(name, _colonIndex, _uri, attributes) {
 *         if (name !== "item") return;
 *         let id = "";
 *         let title = "";
 *         for (let i = 0; i < attributes.count; i++) {
 *           const attrName = attributes.getName(i);
 *           if (attrName === "id") id = attributes.getValue(i);
 *           else if (attrName === "title") title = attributes.getValue(i);
 *         }
 *         emit({ id, title });
 *       },
 *     }),
 *   )
 * ) {
 *   items.push(item);
 * }
 *
 * assertEquals(items, [
 *   { id: "1", title: "First" },
 *   { id: "2", title: "Second" },
 * ]);
 * ```
 */
export async function* parseXmlRecords<T>(
  source: AsyncIterable<string>,
  createCallbacks: (emit: (record: T) => void) => XmlEventCallbacks,
  options: ParseStreamOptions = {},
): AsyncGenerator<T> {
  const buffer: T[] = [];
  const callbacks = createCallbacks((record) => buffer.push(record));
  const { tokenizer, parser } = createXmlPipeline(options, callbacks);

  // Fail-fast contract: a drain-then-throw approach interacts poorly with
  // consumer break, so we propagate immediately and drop the chunk's
  // pending records.
  for await (const chunk of source) {
    tokenizer.process(chunk, parser);
    for (let i = 0; i < buffer.length; i++) yield buffer[i]!;
    buffer.length = 0;
  }
  tokenizer.finalize(parser);
  parser.finalize();
  for (let i = 0; i < buffer.length; i++) yield buffer[i]!;
}

/**
 * Parses an async iterable of XML byte chunks and yields records assembled
 * inside SAX-style event callbacks.
 *
 * This is a convenience wrapper around {@linkcode parseXmlRecords} that
 * handles text decoding via a streaming {@linkcode TextDecoder}. For
 * pre-decoded text streams, use `parseXmlRecords` directly.
 *
 * @typeParam T The type of records yielded by the generator.
 *
 * @param source An async iterable of XML byte chunks (e.g. `response.body`
 * from a `fetch` call).
 *
 * @param createCallbacks Factory invoked once with an `emit` function; returns
 * the SAX-style callbacks that build records and call `emit` per completed
 * record.
 *
 * @param options Parser options forwarded to the underlying tokenizer/parser.
 *
 * @returns An async generator that yields records as the document is parsed.
 *
 * @example Parse items from a byte stream
 * ```ts
 * import { parseXmlRecordsFromBytes } from "@std/xml/parse-records";
 * import { assertEquals } from "@std/assert";
 *
 * const xml = new TextEncoder().encode(
 *   `<feed><item>First</item><item>Second</item></feed>`,
 * );
 *
 * const titles: string[] = [];
 * for await (
 *   const title of parseXmlRecordsFromBytes<string>(
 *     ReadableStream.from([xml]),
 *     (emit) => {
 *       let inside = false;
 *       let text = "";
 *       return {
 *         onStartElement(name) {
 *           if (name === "item") {
 *             inside = true;
 *             text = "";
 *           }
 *         },
 *         onText(t) {
 *           if (inside) text += t;
 *         },
 *         onEndElement(name) {
 *           if (name === "item") {
 *             emit(text);
 *             inside = false;
 *           }
 *         },
 *       };
 *     },
 *   )
 * ) {
 *   titles.push(title);
 * }
 *
 * assertEquals(titles, ["First", "Second"]);
 * ```
 */
export function parseXmlRecordsFromBytes<T>(
  source: AsyncIterable<Uint8Array>,
  createCallbacks: (emit: (record: T) => void) => XmlEventCallbacks,
  options: ParseStreamOptions = {},
): AsyncGenerator<T> {
  return parseXmlRecords(
    decodeAsyncIterable(source),
    createCallbacks,
    options,
  );
}
