/**
 * @blockcast/fec-worker — Shred (Reed-Solomon) Worker Entry Point
 *
 * Runs inside a DedicatedWorkerGlobalScope. Owns the batch-drain: reads raw
 * shred frames (either from a transferred `ReadableStream` or via `feedShred`),
 * parses the wire header, drives the `ShredFecBuffer` WASM decoder, and posts
 * only recovered FEC sets (plus a coalesced stats heartbeat) back to the main
 * thread. This moves the datagram read loop + wire parse + WASM FEC entirely
 * off the main thread, so the page does no `read()`-per-packet work.
 *
 * WASM bindings resolve via:
 *   1. (self as any).__SHRED_WASM_BINDINGS__ — injected by a loader/importmap
 *      or by tests (mirrors the RaptorQ worker's __MMT_WASM_BINDINGS__ contract)
 *   2. dynamic import of config.wasmUrl (wasm-bindgen ESM: `await default()`
 *      then `new ShredFecBuffer(...)`)
 *
 * This file must NOT import DOM globals incompatible with Worker scope.
 */

import type {
	ShredWorkerCommand,
	ShredWorkerConfig,
	ShredWorkerEvent,
	ShredWorkerStats,
	ShredWorkerErrorCode,
	RecoveredMeta,
	SlotSample,
} from "./shred-worker-types.js";
import {
	SHRED_WIRE_VERSION,
	SHRED_WIRE_VERSION_FULL,
	SHRED_HEADER_LEN,
	SHRED_FLAG_IS_CODING,
	SHRED_FLAG_DATA_COMPLETE,
} from "./shred-worker-types.js";
import { toOwnedArrayBuffer } from "./transfer.js";

// --- WASM binding types (from the shred `mmt_wasm` build) -------------------
// Two decoder classes exist, selected the same way the player does
// (`ShredFecBuffer ?? ShredReassembler`) and dispatched by add_shred arity:
//   - ShredFecBuffer   : full Reed-Solomon path, 8-arg add_shred
//       (slot, fecSetIndex, localIndex, numData, numCoding, isCoding, payload, lastInSet)
//   - ShredReassembler : legacy data-only, 5-arg add_shred
//       (slot, fecSetIndex, localIndex, payload, lastInSet); coding shreds ignored
/**
 * A recovered FEC set that carries its `(slot, fecSetIndex)` identity, as
 * returned by `ShredFecBuffer.try_recover_all_with_identity`. wasm-bindgen
 * getters surface `slot` as a BigInt and `fec_set_index` as a number; `payload`
 * is a fresh owned copy (safe to read after `free()`). The object is WASM-owned
 * — call `free()` once consumed.
 */
interface RecoveredSetWasm {
	readonly slot: bigint;
	readonly fec_set_index: number;
	readonly payload: Uint8Array;
	free?(): void;
}

interface ShredDecoderWasm {
	// Loose signature so both arities type-check; the caller branches on
	// `add_shred.length` (wasm-bindgen preserves declared arity).
	add_shred(...args: unknown[]): Uint8Array | null | undefined;
	/** Drain timed-out FEC sets the direct path didn't complete (RS path only). */
	try_recover_all?(nowMs: number): Uint8Array | null | undefined;
	/**
	 * Identity-carrying drain (ShredFecBuffer only): same backlog as
	 * `try_recover_all` but the result keeps its `(slot, fecSetIndex)`, so the
	 * worker can attribute the real send_ts recorded at ingest to a set that
	 * completes on the recover timer rather than inline. Absent on the legacy
	 * reassembler — the worker falls back to `try_recover_all` there.
	 */
	try_recover_all_with_identity?(nowMs: number): RecoveredSetWasm | null | undefined;
	pending_fec_sets?(): number;
	free?(): void;
}

type ShredDecoderCtor = new (maxSlots: bigint, symbolSize: number) => ShredDecoderWasm;

interface ShredWasmBindings {
	/** wasm-bindgen ESM default init (absent when bindings are pre-inited). */
	default?: () => Promise<unknown>;
	ShredFecBuffer?: ShredDecoderCtor;
	ShredReassembler?: ShredDecoderCtor;
}

/**
 * Dispatch one shred to the decoder, mirroring the player's `callAddShred`:
 * ShredFecBuffer (>=7-arg) gets the full RS call; the legacy 5-arg reassembler
 * gets data shreds only (coding shreds are dropped, it can't use them).
 */
function callAddShred(
	dec: ShredDecoderWasm,
	slot: bigint,
	fecSetIndex: number,
	localIndex: number,
	numData: number,
	numCoding: number,
	isCoding: boolean,
	payload: Uint8Array,
	lastInSet: boolean,
): Uint8Array | null | undefined {
	if (dec.add_shred.length >= 7) {
		return dec.add_shred(slot, fecSetIndex, localIndex, numData, numCoding, isCoding, payload, lastInSet);
	}
	if (!isCoding) return dec.add_shred(slot, fecSetIndex, localIndex, payload, lastInSet);
	return null;
}

// --- Worker state -----------------------------------------------------------

let decoder: ShredDecoderWasm | null = null;
// configure() generation — guards against re-entrant configure() races: a later
// call supersedes an earlier one still awaiting the WASM load, so the loser bails
// instead of arming a second set of timers (which would orphan the winner's — a
// leak that outlives dispose()).
let configureGen = 0;
let statsTimer: ReturnType<typeof setInterval> | null = null;
let recoverTimer: ReturnType<typeof setInterval> | null = null;
let activeStreamReader: ReadableStreamDefaultReader<Uint8Array> | null = null;
let disposed = false;

// Slot tracking for main-thread stats (turbine estimator, latency percentiles).
// The worker is the single parser; it hands the main thread first-shred-per-slot
// samples aggregated into the stats heartbeat, never per-packet.
let maxSlot = 0n;
let lastSlotSampled: bigint | null = null;
let slotSamples: SlotSample[] = [];
/** Cap the per-heartbeat sample batch; ~400ms slots means this is rarely hit. */
const SLOT_SAMPLE_CAP = 256;

// Per-FEC-set send timestamp, recorded at ingest (`(slot, fecSetIndex)` →
// first-seen sendTsUs). The loss-recovery drain returns a set that completed on
// the recover timer, not inline, so its completing shred's send_ts isn't in
// hand at emit time; without this, every RS-recovered set would carry sendTsUs
// 0n and the "Listener→browser" latency widgets would stay empty whenever loss
// (the common case over a QUIC datagram path) makes recovery — not inline
// completion — the dominant path.
//
// Key is a bigint pack `(slot << 32) | fecSetIndex` (fecSetIndex is u32) rather
// than a template string — `recordSetSendTs` runs on every ingested shred (a
// documented per-packet hot path), so this avoids a per-frame string alloc.
const setSendTs = new Map<bigint, bigint>();
/**
 * Bound the map for sets that never complete (shreds dropped before enough of a
 * set arrived to recover). Entries are deleted the moment a set completes
 * (either path), so this cap is only reached under sustained un-recoverable
 * loss; ~a few sets/slot means 4096 covers minutes of backlog.
 */
const SET_TS_CAP = 4096;

function setKey(slot: bigint, fecSetIndex: number): bigint {
	return (slot << 32n) | BigInt(fecSetIndex >>> 0);
}

/** Record the first send_ts seen for a set so the recover path can attribute it. */
function recordSetSendTs(slot: bigint, fecSetIndex: number, sendTsUs: bigint): void {
	if (sendTsUs === 0n) return; // unstamped feed — nothing to attribute
	const key = setKey(slot, fecSetIndex);
	if (setSendTs.has(key)) return; // keep the first-seen stamp (matches slotSamples)
	// At cap, reject the NEW entry rather than evicting the oldest — matching the
	// flood-resistance policy of this package's `enqueuePendingVerify`
	// (fec-worker.ts). The oldest pending sets are the ones closest to completing
	// via the recover timer, so evict-oldest would drop attribution exactly when
	// it matters and is attacker-triggerable via a flood of never-completing sets.
	// A rejected set simply falls back to sendTsUs 0n if it later recovers.
	if (setSendTs.size >= SET_TS_CAP) return;
	setSendTs.set(key, sendTsUs);
}

const stats: ShredWorkerStats = {
	shredsRx: 0,
	sourceShreds: 0,
	repairShreds: 0,
	fecSetsCompleted: 0,
	bytesReassembled: 0,
	fecSetsRecovered: 0,
	pendingFecSets: -1,
	malformed: 0,
	versionMismatch: 0,
	currentSlot: "0",
	slotSamples: [],
};

// --- Emit helpers -----------------------------------------------------------

function post(event: ShredWorkerEvent, transfer: Transferable[] = []): void {
	(self as unknown as DedicatedWorkerGlobalScope).postMessage(event, transfer);
}

function postError(message: string, code?: ShredWorkerErrorCode): void {
	post({ type: "error", message, code });
}

function emitRecovered(view: Uint8Array, meta: RecoveredMeta): void {
	// Copy out of WASM-owned memory into a fresh JS buffer before transfer.
	const buf = toOwnedArrayBuffer(view);
	stats.fecSetsCompleted++;
	stats.bytesReassembled += buf.byteLength;
	if (meta.recovered) stats.fecSetsRecovered++;
	post({ type: "recovered", data: buf, meta }, [buf]);
}

function postStats(): void {
	if (decoder && typeof decoder.pending_fec_sets === "function") {
		try {
			stats.pendingFecSets = decoder.pending_fec_sets();
		} catch {
			/* pending count is best-effort */
		}
	}
	stats.currentSlot = maxSlot.toString();
	// Drain the slot-sample batch into this heartbeat; live counters keep mutating.
	const drainedSamples = slotSamples;
	slotSamples = [];
	post({ type: "stats", stats: { ...stats, slotSamples: drainedSamples } });
}

// --- Wire parse + decode ----------------------------------------------------

/**
 * The v3 erasure-shard view of a v4 body (a full canonical shred), or null when
 * the body is not a shred the forwarder could have emitted. Mirrors the
 * forwarder's `parse_shred()` checks and `erasure_shard()` slice exactly, so a
 * v4 frame reaches `add_shred` byte-identical to its v3 twin. A view, not a
 * copy: the hot path allocates nothing extra for v4.
 */
function erasureShardOfFullShred(shred: Uint8Array, isCoding: boolean): Uint8Array | null {
	if (shred.byteLength !== (isCoding ? 1228 : 1203)) return null;
	const variant = shred[64];
	const family = variant & 0xf0;
	const variantIsCoding = family === 0x60 || family === 0x70;
	const variantIsData = family === 0x90 || family === 0xb0;
	if (isCoding ? !variantIsCoding : !variantIsData) return null;
	const resigned = family === 0xb0 || family === 0x70;
	const trailer = 32 + (variant & 0x0f) * 20 + (resigned ? 64 : 0);
	const start = isCoding ? 89 : 64;
	const end = shred.byteLength - trailer;
	if (end <= start) return null;
	return shred.subarray(start, end);
}

/**
 * Parse one raw shred frame and feed the decoder. Mirrors the deployed player's
 * `onShredBytes` (v3 wire, version-guarded) so the off-thread path is
 * bit-identical to the proven on-thread path, and accepts v4 by reducing its
 * full shred to that same v3 erasure shard. Emits a `recovered` event when a
 * set completes directly.
 */
function ingestFrame(frame: Uint8Array): void {
	if (frame.byteLength < SHRED_HEADER_LEN) {
		stats.malformed++;
		return;
	}
	const dv = new DataView(frame.buffer, frame.byteOffset, frame.byteLength);
	const version = dv.getUint8(0);
	if (version !== SHRED_WIRE_VERSION && version !== SHRED_WIRE_VERSION_FULL) {
		// Forwarder/worker wire-version skew — drop rather than misframe every shred.
		stats.versionMismatch++;
		return;
	}
	const slot = dv.getBigUint64(1, true);
	const fecSetIndex = dv.getUint32(9, true);
	const localIndex = dv.getUint32(13, true);
	const flags = dv.getUint8(17);
	const dataComplete = (flags & SHRED_FLAG_DATA_COMPLETE) !== 0; // last data shred in set
	const isCoding = (flags & SHRED_FLAG_IS_CODING) !== 0;
	const numData = dv.getUint8(18);
	const numCoding = dv.getUint8(19);
	const sendTsUs = dv.getBigUint64(20, true);
	const body = frame.subarray(SHRED_HEADER_LEN);
	const payload =
		version === SHRED_WIRE_VERSION ? body : erasureShardOfFullShred(body, isCoding);
	if (payload === null) {
		stats.malformed++;
		return;
	}

	stats.shredsRx++;
	if (isCoding) stats.repairShreds++;
	else stats.sourceShreds++;

	// Slot tracking for main-thread stats. Sample the first shred of each new
	// slot (send timestamp + size) so the page can run the turbine-leg estimator
	// and latency percentiles without re-parsing the wire header itself.
	if (slot > maxSlot) maxSlot = slot;
	if (lastSlotSampled === null || slot !== lastSlotSampled) {
		lastSlotSampled = slot;
		if (slotSamples.length < SLOT_SAMPLE_CAP) {
			slotSamples.push({ slot, sendTsUs, bytes: frame.byteLength });
		}
	}

	// Record this set's send_ts so the recover-timer path (which completes a set
	// without its completing shred in hand) can attribute a real timestamp.
	recordSetSendTs(slot, fecSetIndex, sendTsUs);

	if (!decoder) return;

	// DATA_COMPLETE is passed as `lastInSet`: the WASM uses it to infer numData
	// for a set that received only data shreds (no coding).
	let result: Uint8Array | null | undefined;
	try {
		result = callAddShred(
			decoder,
			slot,
			fecSetIndex,
			localIndex,
			numData,
			numCoding,
			isCoding,
			payload,
			dataComplete,
		);
	} catch (err) {
		postError(`add_shred threw: ${errString(err)}`, "worker-runtime");
		return;
	}
	if (result && result.byteLength > 0) {
		// Set completed inline — its completing shred's send_ts is exact; drop the
		// ingest record so it can't leak or be re-attributed by the recover path.
		setSendTs.delete(setKey(slot, fecSetIndex));
		emitRecovered(result, { slot, fecSetIndex, sendTsUs, recovered: false });
	}
}

/** Run one loss-recovery drain and emit any late-recovered set. */
function recoverAll(nowMs: number): void {
	if (!decoder) return;

	// Preferred path: the identity-carrying drain (ShredFecBuffer) keeps the
	// recovered set's `(slot, fecSetIndex)`, so we can attribute the real send_ts
	// recorded at ingest — otherwise a loss-recovered set (the common case over a
	// lossy datagram path) would carry sendTsUs 0n and the latency widgets would
	// never populate despite valid stamps on the wire.
	if (typeof decoder.try_recover_all_with_identity === "function") {
		let set: RecoveredSetWasm | null | undefined;
		try {
			set = decoder.try_recover_all_with_identity(nowMs);
		} catch (err) {
			postError(`try_recover_all_with_identity threw: ${errString(err)}`, "worker-runtime");
			return;
		}
		if (!set) return;
		// Read the WASM-owned fields before freeing; `payload` is a fresh copy.
		const slot = set.slot;
		const fecSetIndex = set.fec_set_index;
		const payload = set.payload;
		// Best-effort free, matching every other fallible WASM call in this file:
		// recoverAll runs from an unguarded setInterval / message-dispatch branch,
		// so a throwing free() must not propagate and silently drop the set.
		try {
			set.free?.();
		} catch {
			/* best-effort */
		}
		// Clear the map entry on ANY exit for this completed key (empty payload
		// included), so a set can't linger until FIFO eviction.
		const key = setKey(slot, fecSetIndex);
		// sendTsUs here is the set's FIRST-seen shred stamp (recordSetSendTs), vs the
		// inline path's completing-shred stamp; a set's shreds are sent µs apart, so
		// both are equivalent baselines for the "Listener→browser" percentiles.
		const sendTsUs = setSendTs.get(key) ?? 0n;
		setSendTs.delete(key);
		if (!payload || payload.byteLength === 0) return;
		emitRecovered(payload, { slot, fecSetIndex, sendTsUs, recovered: true });
		return;
	}

	// Legacy reassembler: payload-only drain, no set identity available.
	if (typeof decoder.try_recover_all !== "function") return;
	let result: Uint8Array | null | undefined;
	try {
		result = decoder.try_recover_all(nowMs);
	} catch (err) {
		postError(`try_recover_all threw: ${errString(err)}`, "worker-runtime");
		return;
	}
	if (result && result.byteLength > 0) {
		// slot / fecSetIndex / sendTs are not attributable from a bulk drain.
		emitRecovered(result, {
			slot: 0n,
			fecSetIndex: 0,
			sendTsUs: 0n,
			recovered: true,
		});
	}
}

// --- Stream batch-drain -----------------------------------------------------

/**
 * Batch-drain a transferred stream of raw shred frames. The read loop lives
 * here (not on the main thread), so the page never pays a `read()` per packet;
 * only recovered sets and the coalesced stats heartbeat cross back.
 */
async function drainStream(readable: ReadableStream<Uint8Array>): Promise<void> {
	// Replace any prior stream — one active source at a time.
	await cancelActiveStream();
	const reader = readable.getReader();
	activeStreamReader = reader;
	try {
		for (;;) {
			const { value, done } = await reader.read();
			if (done || disposed) break;
			if (value) ingestFrame(value);
		}
	} catch (err) {
		if (!disposed) postError(`stream read failed: ${errString(err)}`, "stream-read-failed");
	} finally {
		if (activeStreamReader === reader) activeStreamReader = null;
		try {
			reader.releaseLock();
		} catch {
			/* already released */
		}
	}
}

async function cancelActiveStream(): Promise<void> {
	const reader = activeStreamReader;
	if (!reader) return;
	activeStreamReader = null;
	try {
		await reader.cancel();
	} catch {
		/* stream may already be closed */
	}
	try {
		reader.releaseLock();
	} catch {
		/* already released */
	}
}

// --- WASM load --------------------------------------------------------------

/** Select the decoder class the same way the player does: RS first, legacy fallback. */
function pickDecoderCtor(b: ShredWasmBindings): ShredDecoderCtor | null {
	if (typeof b.ShredFecBuffer === "function") return b.ShredFecBuffer;
	if (typeof b.ShredReassembler === "function") return b.ShredReassembler;
	return null;
}

async function loadBindings(cfg: ShredWorkerConfig): Promise<ShredWasmBindings> {
	const injected = (self as unknown as { __SHRED_WASM_BINDINGS__?: ShredWasmBindings })
		.__SHRED_WASM_BINDINGS__;
	if (injected && pickDecoderCtor(injected)) {
		return injected;
	}
	if (!cfg.wasmUrl) {
		throw new Error(
			"no shred WASM: set self.__SHRED_WASM_BINDINGS__ = { ShredFecBuffer } or pass config.wasmUrl",
		);
	}
	const mod = (await import(/* @vite-ignore */ cfg.wasmUrl)) as ShredWasmBindings;
	if (typeof mod.default === "function") await mod.default();
	if (!pickDecoderCtor(mod)) {
		throw new Error(
			`shred WASM at ${cfg.wasmUrl} exports neither ShredFecBuffer nor ShredReassembler`,
		);
	}
	return mod;
}

async function configure(cfg: ShredWorkerConfig): Promise<void> {
	const gen = ++configureGen;
	disposed = false;
	teardownTimers();
	await cancelActiveStream();
	if (decoder?.free) {
		try {
			decoder.free();
		} catch {
			/* best-effort */
		}
	}
	decoder = null;
	setSendTs.clear(); // stale per-set stamps must not survive a reconfigure

	let bindings: ShredWasmBindings;
	try {
		bindings = await loadBindings(cfg);
	} catch (err) {
		if (gen === configureGen) postError(errString(err), "wasm-load-failed");
		return;
	}
	if (gen !== configureGen) return; // superseded by a newer configure() while loading
	const Ctor = pickDecoderCtor(bindings);
	if (!Ctor) {
		postError("no shred decoder constructor after load", "wasm-bindings-missing");
		return;
	}
	try {
		decoder = new Ctor(BigInt(cfg.maxSlots), cfg.symbolSize);
	} catch (err) {
		postError(`shred decoder ctor threw: ${errString(err)}`, "wasm-load-failed");
		return;
	}

	// Clear any timer a racing continuation may have armed between the await and
	// here, so exactly one stats/recover interval is live.
	teardownTimers();
	const statsMs = cfg.statsIntervalMs ?? 250;
	if (statsMs > 0) statsTimer = setInterval(postStats, statsMs);
	const recoverMs = cfg.recoverIntervalMs ?? 50;
	if (recoverMs > 0) recoverTimer = setInterval(() => recoverAll(Date.now()), recoverMs);
}

function teardownTimers(): void {
	if (statsTimer !== null) {
		clearInterval(statsTimer);
		statsTimer = null;
	}
	if (recoverTimer !== null) {
		clearInterval(recoverTimer);
		recoverTimer = null;
	}
}

async function dispose(): Promise<void> {
	disposed = true;
	teardownTimers();
	await cancelActiveStream();
	if (decoder?.free) {
		try {
			decoder.free();
		} catch {
			/* best-effort */
		}
	}
	decoder = null;
	setSendTs.clear();
}

// --- Message dispatch -------------------------------------------------------

self.onmessage = (e: MessageEvent<ShredWorkerCommand>): void => {
	const cmd = e.data;
	switch (cmd.type) {
		case "configure":
			void configure(cmd.config).catch((err) =>
				postError(errString(err), "wasm-load-failed"),
			);
			break;
		case "attachStream":
			if (!decoder) {
				postError("attachStream before configure", "not-configured");
				// Drop the transferred stream so it doesn't leak.
				void cancelStreamHandle(cmd.readable);
				break;
			}
			void drainStream(cmd.readable);
			break;
		case "feedShred":
			if (!decoder) {
				postError("feedShred before configure", "not-configured");
				break;
			}
			ingestFrame(new Uint8Array(cmd.frame));
			break;
		case "recoverAll":
			recoverAll(cmd.nowMs);
			break;
		case "dispose":
			void dispose();
			break;
		default: {
			const _exhaustive: never = cmd;
			postError(`unknown command: ${JSON.stringify(_exhaustive)}`, "unknown-command");
		}
	}
};

self.onmessageerror = (): void => {
	postError("failed to deserialize worker message", "worker-message");
};

async function cancelStreamHandle(readable: ReadableStream<Uint8Array>): Promise<void> {
	try {
		await readable.cancel();
	} catch {
		/* best-effort */
	}
}

function errString(err: unknown): string {
	if (err instanceof Error) return err.message;
	return String(err);
}