import * as announce from "../announced.ts";
import * as broadcast from "../broadcast.ts";
import { BroadcastCache } from "../consume.ts";
import { error, reason } from "../error.ts";
import * as netGroup from "../group.ts";
import * as Path from "../path.ts";
import type { Reader, Stream } from "../stream.ts";
import { type Timescale, Timestamp } from "../time.ts";
import type * as track from "../track.ts";
import { withTimeout } from "../util/timeout.ts";
import type { Session } from "./adapter.ts";
import { TrackAliases } from "./aliases.ts";
import { Frame, type Group as GroupMessage } from "./object.ts";
import { type Publish, PublishError } from "./publish.ts";
import { type PublishNamespace, PublishNamespaceError, PublishNamespaceOk } from "./publish_namespace.ts";
import { RequestError, RequestOk } from "./request.ts";
import { Subscribe, SubscribeError, SubscribeOk, Unsubscribe } from "./subscribe.ts";
import {
	PublishBlocked,
	SubscribeNamespace,
	SubscribeNamespaceEntry,
	SubscribeNamespaceEntryDone,
	SubscribeNamespaceLegacy,
	SubscribeNamespaceOk,
	UnsubscribeNamespace,
} from "./subscribe_namespace.ts";
import { Version } from "./version.ts";

// Bound on how long stream-open plus SUBSCRIBE_OK may take. Browsers cap
// concurrent QUIC streams (Chrome ~100); past the cap openBi() silently
// blocks. The timeout turns that into a clear error.
const SUBSCRIBE_OK_TIMEOUT_MS = 10_000;

// Out-parameter for #openSubscribe: lets the caller observe partial progress
// (stream opened, trackAlias registered) so it can clean up on timeout even
// before the setup promise settles.
type SubscribeSetupState = {
	stream?: Stream;
	registeredAlias?: bigint;
};

/**
 * Handles subscribing to broadcasts using moq-transport protocol.
 * Uses the stream-per-request pattern (real bidi streams for v17, virtual for v14-v16).
 *
 * @internal
 */
export class Subscriber {
	#session: Session;

	// Publisher-chosen aliases used by incoming group streams.
	#aliases = new TrackAliases<track.Producer>();

	// Units for each track's object Timestamps, from the TIMESCALE Track Property in
	// SUBSCRIBE_OK. A track missing from this map declared no timeline, so the publisher
	// opted out of timestamps and its frames are stamped on arrival instead.
	#timescales = new Map<bigint, Timescale>();

	// Dedup consumed broadcasts per path: repeat consume() calls share one subscription.
	#consumes = new BroadcastCache();

	// Any currently active announcements.
	#announced = new Set<Path.Valid>();

	// Any consumers that want each new announcement.
	#announcedConsumers = new Set<announce.Producer>();

	/**
	 * Creates a new Subscriber instance.
	 * @param session - The session abstraction for bidi streams and request IDs
	 *
	 * @internal
	 */
	constructor(session: Session) {
		this.#session = session;
	}

	/**
	 * Gets an announced reader for the specified prefix.
	 */
	announced(prefix = Path.empty()): announce.Consumer {
		const announced = new announce.Producer(prefix);
		for (const active of this.#announced) {
			const suffix = Path.stripPrefix(prefix, active);
			if (suffix === null) continue;
			announced.append({ path: suffix, active: true });
		}
		this.#announcedConsumers.add(announced);

		void this.#runAnnounced(announced, prefix).finally(() => {
			this.#announcedConsumers.delete(announced);
			announced.close();
		});

		return announced.consume();
	}

	async #runAnnounced(announced: announce.Producer, prefix: Path.Valid) {
		const version = this.#session.version;

		// v14/v15: SubscribeNamespace on control stream (via adapter virtual stream)
		// v16+: SubscribeNamespace on its own real bidi stream

		const requestId = await this.#session.nextRequestId();
		if (requestId === undefined) return;

		try {
			// v16: use a real bidi stream (not virtual control stream)
			const stream =
				version === Version.DRAFT_16 && this.#session.openNativeBi
					? await this.#session.openNativeBi()
					: await this.#session.openBi();

			try {
				// Draft-18+ uses SUBSCRIBE_NAMESPACE (0x50); earlier drafts use the
				// legacy 0x11 message with a Subscribe Options field.
				if (
					version === Version.DRAFT_14 ||
					version === Version.DRAFT_15 ||
					version === Version.DRAFT_16 ||
					version === Version.DRAFT_17
				) {
					await stream.writer.u53(SubscribeNamespaceLegacy.id);
					await new SubscribeNamespaceLegacy({ namespace: prefix, requestId }).encode(stream.writer, version);
				} else {
					await stream.writer.u53(SubscribeNamespace.id);
					await new SubscribeNamespace({ namespace: prefix, requestId }).encode(stream.writer, version);
				}
				console.debug(`subscribe_namespace written: requestId=${requestId}`);

				// Read response
				const respTypeId = await stream.reader.u53();
				if (respTypeId === RequestOk.id) {
					await RequestOk.decode(stream.reader, version);
				} else if (respTypeId === SubscribeNamespaceOk.id) {
					// v14: SubscribeNamespaceOk
					const size = await stream.reader.u16();
					await stream.reader.read(size);
				} else {
					throw new Error(`SubscribeNamespace rejected: typeId=0x${respTypeId.toString(16)}`);
				}

				// Loop reading Namespace/NamespaceDone entries
				const readLoop = (async () => {
					for (;;) {
						const done = await stream.reader.done();
						if (done) break;

						const msgType = await stream.reader.u53();
						if (msgType === SubscribeNamespaceEntry.id) {
							const entry = await SubscribeNamespaceEntry.decode(stream.reader, version);
							const path = Path.join(prefix, entry.suffix);
							console.debug(`announced: broadcast=${path} active=true`);

							this.#announced.add(path);
							for (const consumer of this.#announcedConsumers) {
								const suffix = Path.stripPrefix(consumer.prefix, path);
								if (suffix === null) continue;
								consumer.append({ path: suffix, active: true });
							}
						} else if (msgType === SubscribeNamespaceEntryDone.id) {
							const entry = await SubscribeNamespaceEntryDone.decode(stream.reader, version);
							const path = Path.join(prefix, entry.suffix);
							console.debug(`announced: broadcast=${path} active=false`);

							this.#announced.delete(path);
							this.#consumes.evict(path);
							for (const consumer of this.#announcedConsumers) {
								const suffix = Path.stripPrefix(consumer.prefix, path);
								if (suffix === null) continue;
								consumer.append({ path: suffix, active: false });
							}
						} else if (msgType === PublishBlocked.id && version === Version.DRAFT_17) {
							const blocked = await PublishBlocked.decode(stream.reader, version);
							console.debug(`publish_blocked: suffix=${blocked.suffix} track=${blocked.trackName}`);
						} else {
							throw new Error(
								`unexpected message on subscribe_namespace stream: 0x${msgType.toString(16)}`,
							);
						}
					}
				})();

				// Wait for either the read loop or the announced to close
				await Promise.race([readLoop, announced.closed]);

				// For v14/v15: send UnsubscribeNamespace before closing
				if (version === Version.DRAFT_14 || version === Version.DRAFT_15) {
					try {
						await stream.writer.u53(UnsubscribeNamespace.id);
						const unsub = new UnsubscribeNamespace({ requestId });
						await unsub.encode(stream.writer, version);
					} catch {
						// Stream might already be closed
					}
				}

				stream.close();
			} catch (err) {
				stream.abort(error(err));
				throw err;
			}
		} catch (err: unknown) {
			const e = error(err);
			console.warn(`subscribe_namespace error: ${reason(e)}`);

			// Abort the stream rather than letting the caller's `finally` close it cleanly.
			// A rejected namespace subscription is a failure, and a consumer that can't tell
			// it from "nothing is published under this prefix" waits forever on a broadcast
			// that will never be announced. Matches the lite subscriber.
			announced.close(e);
		}
	}

	/**
	 * Consumes a broadcast from the connection.
	 *
	 * Deduplicated per path: repeat calls for the same still-live path share one reference-counted
	 * broadcast (and one upstream subscription). The shared broadcast closes once every caller has
	 * closed its handle, so callers close normally.
	 */
	consume(path: Path.Valid): broadcast.Consumer {
		return this.#consumes.get(path) ?? this.#consumes.insert(path, this.#createConsume(path));
	}

	#createConsume(path: Path.Valid): broadcast.Consumer {
		// moq-transport has no one-shot group fetch; ConsumeBroadcast rejects it. Track info
		// is resolved by the subscribe path (the inherited resolveTrackInfo).
		const consumer = new ConsumeBroadcast();

		void (async () => {
			for (;;) {
				const request = await consumer.requested();
				if (!request) break;
				void this.#runSubscribe(path, request);
			}
		})();

		return consumer;
	}

	async #runSubscribe(broadcast: Path.Valid, request: track.Request) {
		const version = this.#session.version;
		const requestId = await this.#session.nextRequestId();
		if (requestId === undefined) {
			request.reject(new Error("session closed"));
			return;
		}

		console.debug(`subscribe start: id=${requestId} broadcast=${broadcast} track=${request.name}`);

		// IETF negotiates group order in SUBSCRIBE_OK; this implementation only
		// supports descending (newest-first), so commit ordered: false. (There's no
		// per-frame timescale, so the rest stay at their defaults.) This
		// resolves the consumer's track.info() and gives us the write side that
		// incoming object streams are routed into.
		const producer = request.accept({ ordered: false });

		// Open the stream and wait for SUBSCRIBE_OK under a timeout. State
		// flows back via `state` so the timeout path can clean up the stream
		// and any registration if setup eventually finishes.
		const state: SubscribeSetupState = {};
		const setup = this.#openSubscribe(state, broadcast, request, producer, requestId);

		let stream: Stream;
		let trackAlias: bigint;
		try {
			const result = await withTimeout(
				setup,
				SUBSCRIBE_OK_TIMEOUT_MS,
				`subscribe timed out after ${SUBSCRIBE_OK_TIMEOUT_MS}ms waiting for SUBSCRIBE_OK (browser stream limit reached?)`,
			);
			stream = result.stream;
			trackAlias = result.alias;
			console.debug(`subscribe ok: id=${requestId} broadcast=${broadcast} track=${request.name}`);
		} catch (err) {
			const e = error(err);
			producer.close(e);
			console.warn(
				`subscribe error: id=${requestId} broadcast=${broadcast} track=${request.name} error=${reason(e)}`,
			);
			// If setup eventually settles after the timeout, abort the stream
			// and drop any registration so we don't leak. Cover both branches:
			// setup may resolve late, or reject (e.g. SUBSCRIBE error) after the
			// stream is already open.
			const cleanup = () => {
				if (state.registeredAlias !== undefined) this.#aliases.delete(state.registeredAlias, producer);
				state.stream?.abort(e);
			};
			setup.then(cleanup, cleanup);
			return;
		}

		try {
			// Terminal conditions settle at most once (stream close = PublishDone, track close =
			// local unsubscribe); race them once so the demand loop doesn't re-subscribe each pass.
			const done = Promise.race([stream.reader.closed, producer.closed]);

			// Serve until a terminal condition fires or the last local subscriber leaves. The unused
			// wake is level-triggered: re-check demand so a subscriber that returns before we tear
			// down resumes on the same stream.
			const idle = Symbol("idle");
			for (;;) {
				const reason = await Promise.race([done, producer.unused().then(() => idle)]);
				if (reason === idle && producer.closed.peek() === undefined && producer.used.peek()) continue;
				break;
			}

			// For v14-v16: send Unsubscribe before closing (removed in v17+)
			if (version === Version.DRAFT_14 || version === Version.DRAFT_15 || version === Version.DRAFT_16) {
				try {
					await stream.writer.u53(Unsubscribe.id);
					const unsub = new Unsubscribe({ requestId });
					await unsub.encode(stream.writer, version);
				} catch {
					// Stream might already be closed
				}
			}

			producer.close();
			stream.close();
			console.debug(`subscribe close: id=${requestId} broadcast=${broadcast} track=${request.name}`);
		} catch (err) {
			const e = error(err);
			producer.close(e);
			stream.abort(e);
			console.warn(
				`subscribe error: id=${requestId} broadcast=${broadcast} track=${request.name} error=${reason(e)}`,
			);
		} finally {
			this.#aliases.delete(trackAlias, producer);
			this.#timescales.delete(trackAlias);
		}
	}

	// Opens the subscribe stream, sends SUBSCRIBE, and reads the response.
	// `state` is populated as soon as the stream opens and again when the
	// trackAlias is registered, so the caller can clean both up on timeout
	// even before this promise settles.
	async #openSubscribe(
		state: SubscribeSetupState,
		broadcast: Path.Valid,
		request: track.Request,
		producer: track.Producer,
		requestId: bigint,
	): Promise<{ stream: Stream; alias: bigint }> {
		const version = this.#session.version;

		state.stream = await this.#session.openBi();

		await state.stream.writer.u53(Subscribe.id);
		const msg = new Subscribe({
			requestId,
			trackNamespace: broadcast,
			trackName: request.name,
			subscriberPriority: request.priority,
		});
		await msg.encode(state.stream.writer, version);
		console.debug(`subscribe written: id=${requestId} broadcast=${broadcast} track=${request.name}`);

		const respTypeId = await state.stream.reader.u53();
		if (respTypeId !== SubscribeOk.id) {
			let reasonPhrase = "unknown error";
			try {
				if (respTypeId === RequestError.id) {
					const err =
						version === Version.DRAFT_14
							? await SubscribeError.decode(state.stream.reader, version)
							: await RequestError.decode(state.stream.reader, version);
					reasonPhrase = `code=${err.errorCode} reason=${err.reasonPhrase}`;
				}
			} catch {
				// Decoding error response failed, use default message
			}
			throw new Error(`SUBSCRIBE error: ${reasonPhrase}`);
		}

		const ok = await SubscribeOk.decode(state.stream.reader, version);
		try {
			this.#aliases.set(ok.trackAlias, producer);
			if (ok.timescale !== undefined) {
				this.#timescales.set(ok.trackAlias, ok.timescale);
			}
		} catch (err) {
			this.#session.close();
			throw err;
		}
		state.registeredAlias = ok.trackAlias;
		return { stream: state.stream, alias: ok.trackAlias };
	}

	/**
	 * Handles an incoming PUBLISH_NAMESPACE on a bidi stream.
	 * Tracks announced broadcasts and notifies consumers.
	 *
	 * @internal
	 */
	async runPublishNamespace(msg: PublishNamespace, stream: Stream) {
		const version = this.#session.version;
		const path = msg.trackNamespace;

		if (this.#announced.has(path)) {
			console.warn("duplicate PublishNamespace");
			if (version === Version.DRAFT_14) {
				await stream.writer.u53(PublishNamespaceError.id);
				const err = new PublishNamespaceError({
					requestId: msg.requestId,
					errorCode: 409,
					reasonPhrase: "duplicate namespace",
				});
				await err.encode(stream.writer, version);
			} else {
				await stream.writer.u53(RequestError.id);
				const err = new RequestError({
					requestId: version === Version.DRAFT_15 || version === Version.DRAFT_16 ? msg.requestId : undefined,
					errorCode: 409,
					reasonPhrase: "duplicate namespace",
				});
				await err.encode(stream.writer, version);
			}
			stream.close();
			return;
		}

		this.#announced.add(path);

		try {
			// Send OK first. This must complete before notifying consumers,
			// because consumers may trigger Subscribe writes that would
			// interleave with our OK on the control stream.
			if (version === Version.DRAFT_14) {
				await stream.writer.u53(PublishNamespaceOk.id);
				const ok = new PublishNamespaceOk({ requestId: msg.requestId });
				await ok.encode(stream.writer, version);
			} else {
				await stream.writer.u53(RequestOk.id);
				const ok = new RequestOk({
					requestId: version === Version.DRAFT_15 || version === Version.DRAFT_16 ? msg.requestId : undefined,
				});
				await ok.encode(stream.writer, version);
			}

			console.debug(`announced: broadcast=${path} active=true`);

			// Notify consumers after OK is written
			for (const consumer of this.#announcedConsumers) {
				const suffix = Path.stripPrefix(consumer.prefix, path);
				if (suffix === null) continue;
				consumer.append({ path: suffix, active: true });
			}

			// Wait for stream close (= PublishNamespaceDone)
			console.debug(`runPublishNamespace: awaiting stream.reader.closed for ${path}`);
			await stream.reader.closed;
			console.debug(`runPublishNamespace: stream.reader.closed resolved for ${path}`);
		} finally {
			this.#announced.delete(path);
			// The path is gone, so stop sharing its broadcast: a holder outliving the publisher
			// would otherwise hand the dead generation to whoever consumes the path next.
			this.#consumes.evict(path);
			console.debug(`announced: broadcast=${path} active=false`);

			for (const consumer of this.#announcedConsumers) {
				const suffix = Path.stripPrefix(consumer.prefix, path);
				if (suffix === null) continue;
				try {
					consumer.append({ path: suffix, active: false });
				} catch {
					// Consumer already closed, will be cleaned up
				}
			}
		}
	}

	/**
	 * Handles an incoming PUBLISH on a bidi stream.
	 * We don't support reverse publish, so send error.
	 *
	 * @internal
	 */
	async runPublish(msg: Publish, stream: Stream) {
		const version = this.#session.version;

		if (version === Version.DRAFT_14) {
			await stream.writer.u53(PublishError.id);
			const err = new PublishError({
				requestId: msg.requestId,
				errorCode: 500,
				reasonPhrase: "publish not supported",
			});
			await err.encode(stream.writer, version);
		} else {
			await stream.writer.u53(RequestError.id);
			const err = new RequestError({
				requestId: version === Version.DRAFT_15 || version === Version.DRAFT_16 ? msg.requestId : undefined,
				errorCode: 500,
				reasonPhrase: "publish not supported",
			});
			await err.encode(stream.writer, version);
		}
		stream.close();
	}

	/**
	 * Handles an ObjectStream message (group + frames on uni stream).
	 *
	 * @internal
	 */
	async handleGroup(group: GroupMessage, stream: Reader) {
		const producer = new netGroup.Producer(group.groupId);

		if (group.subGroupId !== 0) {
			throw new Error("subgroups are not supported");
		}

		try {
			// The control message establishing this alias can arrive after the data stream.
			const track = await this.#aliases.get(group.trackAlias);

			track.writeGroup(producer);

			for (;;) {
				const done = await Promise.race([stream.done(), producer.closed, track.closed]);
				if (done !== false) break;

				const frame = await Frame.decode(
					stream,
					group.flags,
					this.#timescales.get(group.trackAlias),
					this.#session.version,
				);
				if (frame.payload === undefined) break;

				producer.writeFrame({ payload: frame.payload, timestamp: frame.timestamp ?? Timestamp.now() });
			}

			producer.close();
		} catch (err: unknown) {
			const e = error(err);
			producer.close(e);
			stream.stop(e);
		}
	}
}

/**
 * A broadcast consumed from a moq-transport session. Track info is resolved by the
 * subscribe path (the inherited `resolveTrackInfo`), but the protocol has no one-shot
 * group fetch, so `track.Consumer.fetchGroup()` is rejected.
 */
class ConsumeBroadcast extends broadcast.Consumer {
	// biome-ignore lint/complexity/noUselessConstructor: widens the protected base constructor to public
	constructor(state?: never) {
		super(state);
	}

	// Preserve the subclass when the consume cache shares this broadcast across callers.
	override clone(): ConsumeBroadcast {
		return new ConsumeBroadcast(this.shareState());
	}

	override fetchGroup(): Promise<netGroup.Consumer> {
		return Promise.reject(new Error("fetch group is not supported for moq-transport"));
	}
}
