/**
 * @blockcast/fec-worker — Shred Worker Main-Thread Client
 *
 * Thin wrapper over the shred (Reed-Solomon) worker, mirroring
 * {@link FecWorkerClient}'s shape. Spawns one Worker, speaks the
 * {@link ShredWorkerCommand}/{@link ShredWorkerEvent} protocol, and keeps the
 * datagram read loop + wire parse + WASM FEC off the main thread.
 *
 * Typical wiring for a WebTransport datagram source (fully off-thread):
 *
 *   const client = new ShredWorkerClient(new URL("./shred-fec-worker.js", import.meta.url));
 *   client.onRecovered((data, meta) => renderRecoveredSet(data, meta));
 *   client.configure({ maxSlots: 5000, symbolSize: 512, wasmUrl: "/vendor/fec/mmt_wasm.js" });
 *   client.attachStream(wt.datagrams.readable); // transferred; main does no read() per packet
 *
 * For sources that demux on the main thread (moqtail OBJECT_DATAGRAM callback,
 * IWA multicast bridge), push each raw frame with {@link feedShred} instead.
 *
 * {@link feedShred} is a fire-and-forget `postMessage` with no native
 * backpressure, so a consumer that decodes slower than the ingest rate (a
 * backgrounded/throttled tab, WASM RS falling behind a ~4k shred/s feed) would
 * grow the worker's inbound message queue without limit — the queue lives in the
 * renderer process and has been observed to reach multiple GB, GC-thrashing the
 * page into a wedge. The client bounds it with an in-flight window keyed off the
 * worker's own periodic stats heartbeat (no per-frame ack round-trip); see
 * {@link ShredWorkerClientOptions.maxInflight}.
 */

import type {
	ShredWorkerCommand,
	ShredWorkerConfig,
	ShredWorkerEvent,
	ShredWorkerStats,
	ShredWorkerErrorCode,
	RecoveredMeta,
} from "./shred-worker-types.js";

/** Construction-time options for {@link ShredWorkerClient}. */
export interface ShredWorkerClientOptions {
	/**
	 * Cap on frames posted via {@link ShredWorkerClient.feedShred} but not yet
	 * drained by the worker, before further frames are dropped ("skip-to-live" at
	 * the worker boundary). Bounds the worker's inbound message queue — the sole
	 * unbounded structure on the main-thread-demux (`feedShred`) path.
	 *
	 * The drain signal is the worker's periodic `stats` heartbeat: the monotonic
	 * total of frames it has taken off the queue (`shredsRx + malformed +
	 * versionMismatch`). No per-frame acknowledgement crosses the boundary, so the
	 * bound is heartbeat-granular — size it above one heartbeat's worth of ingest
	 * (default 250ms heartbeat × ~4k frames/s ≈ ~1k frames).
	 *
	 * Default {@link DEFAULT_MAX_INFLIGHT} (8192 frames ≈ ~10 MB at ~1.3 KB/frame).
	 * Set to `0` (or any value `<= 0`) to disable the bound entirely — only do so
	 * for a source with its own backpressure (e.g. a bounded {@link
	 * ShredWorkerClient.attachStream} producer). Note the bound assumes `feedShred`
	 * is the sole ingest path; mixing it with `attachStream` (whose frames also
	 * advance the heartbeat total) makes the window under-count and benignly
	 * disables the drop.
	 */
	maxInflight?: number;
}

/** Default {@link ShredWorkerClientOptions.maxInflight} (~10 MB of queued frames). */
export const DEFAULT_MAX_INFLIGHT = 8192;

export class ShredWorkerClient {
	readonly #worker: Worker;
	#onRecovered: ((data: ArrayBuffer, meta: RecoveredMeta) => void) | null = null;
	#onStats: ((stats: ShredWorkerStats) => void) | null = null;
	#onError: ((message: string, code?: ShredWorkerErrorCode) => void) | null = null;
	#disposed = false;

	/** feedShred in-flight backpressure accounting (see ShredWorkerClientOptions.maxInflight). */
	readonly #maxInflight: number;
	#sent = 0; // frames posted via feedShred (monotonic)
	#processed = 0; // worker-consumed total from the last stats heartbeat (monotonic)
	#droppedBackpressure = 0; // frames dropped by the in-flight guard (monotonic)

	/**
	 * @param workerOrUrl - A URL (a Worker is created internally, `type: module`)
	 *   or a pre-created Worker (for testing via constructor injection).
	 * @param opts - Optional {@link ShredWorkerClientOptions}.
	 */
	constructor(workerOrUrl: Worker | URL, opts: ShredWorkerClientOptions = {}) {
		this.#maxInflight = opts.maxInflight ?? DEFAULT_MAX_INFLIGHT;
		this.#worker =
			workerOrUrl instanceof URL
				? new Worker(workerOrUrl, { type: "module" })
				: workerOrUrl;
		this.#worker.onmessage = (e: MessageEvent<ShredWorkerEvent>) =>
			this.#handleEvent(e.data);
		this.#worker.onerror = (e) =>
			this.#onError?.(formatWorkerError(e), "worker-runtime");
		this.#worker.onmessageerror = () =>
			this.#onError?.("failed to deserialize worker message", "worker-message");
	}

	/** Configure the decoder (WASM load + ShredFecBuffer construction happen in the worker). */
	configure(config: ShredWorkerConfig): void {
		if (this.#disposed) throw new Error("[ShredWorkerClient] disposed");
		this.#post({ type: "configure", config });
	}

	/**
	 * Hand the worker a stream of raw shred frames to batch-drain. The stream is
	 * transferred — the caller must not use it afterward. One active stream at a
	 * time; attaching a new one cancels the previous.
	 */
	attachStream(readable: ReadableStream<Uint8Array>): void {
		if (this.#disposed) return;
		this.#post({ type: "attachStream", readable }, [readable]);
	}

	/**
	 * Feed one already-demuxed raw shred frame. `frame` is transferred zero-copy;
	 * the caller must not access it afterward.
	 *
	 * Dropped (frame NOT transferred, so the caller's buffer stays intact) when the
	 * in-flight window is full — i.e. the worker is more than `maxInflight` frames
	 * behind. This is the memory bound: a slow/wedged consumer costs bounded
	 * skip-to-live drops (see {@link droppedBackpressure}), never an unbounded
	 * renderer-heap blow-up.
	 */
	feedShred(frame: ArrayBuffer): void {
		if (this.#disposed) return;
		if (this.#maxInflight > 0 && this.#sent - this.#processed >= this.#maxInflight) {
			this.#droppedBackpressure++;
			return;
		}
		this.#sent++;
		this.#post({ type: "feedShred", frame }, [frame]);
	}

	/** Explicitly drive loss recovery (in addition to the worker's internal timer). */
	recoverAll(nowMs: number = Date.now()): void {
		if (this.#disposed) return;
		this.#post({ type: "recoverAll", nowMs });
	}

	/** Register callback for recovered/completed FEC sets. `data` is a transferred buffer. */
	onRecovered(cb: (data: ArrayBuffer, meta: RecoveredMeta) => void): void {
		this.#onRecovered = cb;
	}

	/** Register callback for the coalesced stats heartbeat. */
	onStats(cb: (stats: ShredWorkerStats) => void): void {
		this.#onStats = cb;
	}

	/** Register callback for worker errors. */
	onError(cb: (message: string, code?: ShredWorkerErrorCode) => void): void {
		this.#onError = cb;
	}

	/**
	 * Frames dropped by the in-flight backpressure guard since construction
	 * (monotonic). Surface this in the UI so sustained loss is visible rather than
	 * silent — a rising counter means the consumer can't keep pace with ingest.
	 */
	get droppedBackpressure(): number {
		return this.#droppedBackpressure;
	}

	/**
	 * Frames posted via {@link feedShred} but not yet reported drained by the
	 * worker's last stats heartbeat (best-effort, heartbeat-granular; never < 0).
	 */
	get inflight(): number {
		return Math.max(0, this.#sent - this.#processed);
	}

	/** Dispose the worker. Idempotent. */
	dispose(): void {
		if (this.#disposed) return;
		this.#disposed = true;
		try {
			this.#worker.postMessage({ type: "dispose" } satisfies ShredWorkerCommand);
		} catch {
			/* worker may already be gone */
		}
		this.#worker.terminate();
	}

	#post(cmd: ShredWorkerCommand, transfer: Transferable[] = []): void {
		this.#worker.postMessage(cmd, transfer);
	}

	#handleEvent(event: ShredWorkerEvent): void {
		switch (event.type) {
			case "recovered":
				this.#onRecovered?.(event.data, event.meta);
				break;
			case "stats":
				// The worker's monotonic consumed-frame total is the drain signal for
				// the feedShred in-flight guard. shredsRx counts only valid frames;
				// malformed + versionMismatch are consumed off the queue too, so the
				// sum is the true "frames taken off the queue" and keeps the window
				// from inflating (and false-dropping) on a malformed/skewed source.
				this.#processed =
					event.stats.shredsRx + event.stats.malformed + event.stats.versionMismatch;
				this.#onStats?.(event.stats);
				break;
			case "error":
				if (event.code) this.#onError?.(event.message, event.code);
				else this.#onError?.(event.message);
				break;
		}
	}
}

function formatWorkerError(event: ErrorEvent): string {
	const parts = ["Worker error"];
	if (event.message) parts.push(event.message);
	if (event.filename) {
		const loc = [event.filename];
		if (event.lineno) loc.push(String(event.lineno));
		if (event.colno) loc.push(String(event.colno));
		parts.push(`at ${loc.join(":")}`);
	}
	if (event.error instanceof Error && event.error.stack) parts.push(event.error.stack);
	return parts.join(": ");
}