import { type Getter, Signal } from "@moq/signals";
import * as announce from "../announced.ts";
import type * as broadcast from "../broadcast.ts";
import type { Established } from "../connection/established.ts";
import { type Probe, type Stats, transportStats } from "../connection/stats.ts";
import { type Transport, transportOf } from "../connection/transport.ts";
import * as Path from "../path.ts";
import { type Reader, Readers, type Stream } from "../stream.ts";
import { ControlStreamAdapter, NativeSession, type Session } from "./adapter.ts";
import { GoAway } from "./goaway.ts";
import { Group } from "./object.ts";
import { Publish } from "./publish.ts";
import { PublishNamespace } from "./publish_namespace.ts";
import { Publisher } from "./publisher.ts";
import { Subscribe, SubscribeUpdate } from "./subscribe.ts";
import { SubscribeNamespace, SubscribeNamespaceLegacy } from "./subscribe_namespace.ts";
import { Subscriber } from "./subscriber.ts";
import { TrackStatusRequest } from "./track.ts";
import { type IetfVersion, Version, versionName } from "./version.ts";

/**
 * Represents a connection to a MoQ server using moq-transport protocol.
 *
 * @public
 */
export class Connection implements Established {
	// The URL of the connection.
	readonly url: URL;

	// The negotiated protocol version.
	readonly version: string;

	// The wire transport this session runs over.
	readonly transport: Transport;

	/** Whether the relay supports broadcast discovery; see {@link Established.discovery}. */
	readonly discovery: boolean;

	/** moq-transport has no PROBE, so this stays empty; see {@link Established.probe}. */
	readonly probe: Getter<Probe> = new Signal<Probe>({});

	// The established WebTransport session.
	#quic: WebTransport;

	// Session abstraction: adapter for v14-v16, native for v17.
	#session: Session;

	// Module for contributing tracks.
	#publisher: Publisher;

	// Module for distributing tracks.
	#subscriber: Subscriber;

	// Just to avoid logging when `close()` is called.
	#closed = false;

	/**
	 * Creates a new Connection instance.
	 * @param url - The URL of the connection
	 * @param quic - The WebTransport session
	 * @param control - The control/setup stream
	 * @param maxRequestId - The initial max request ID
	 * @param version - The negotiated protocol version
	 *
	 * @internal
	 */
	constructor({
		url,
		quic,
		control,
		maxRequestId,
		version,
		client,
		discovery = true,
	}: {
		url: URL;
		quic: WebTransport;
		control: Stream;
		maxRequestId: bigint;
		version: IetfVersion;
		/** Whether this peer initiated the session, selecting the even request-ID space. */
		client: boolean;
		discovery?: boolean;
	}) {
		this.url = url;
		this.discovery = discovery;
		this.version = versionName(version);
		this.transport = transportOf(quic);
		this.#quic = quic;

		// Two-path dispatch: v14-v16 uses adapter, v17+ uses native bidi streams
		if (version >= Version.DRAFT_17) {
			this.#session = new NativeSession(quic, version, client);
			// v17+: control/setup stream only carries GoAway
			void this.#runGoAway(control, version);
		} else {
			const adapter = new ControlStreamAdapter(quic, control, version, maxRequestId, client);
			this.#session = adapter;
			// Start the adapter read loop (routes control messages to virtual streams)
			void adapter.run().catch((err: unknown) => {
				if (!this.#closed) console.error("adapter error", err);
				this.close();
			});
		}

		this.#publisher = new Publisher(this.#quic, this.#session);
		this.#subscriber = new Subscriber(this.#session);

		void this.#run();
	}

	/** Snapshot the transport's counters; see {@link Established.stats}. */
	async stats(): Promise<Stats> {
		return transportStats(this.#quic);
	}

	/**
	 * Closes the connection.
	 */
	close() {
		if (this.#closed) return;

		this.#closed = true;

		this.#publisher.close();
		this.#session.close();

		try {
			this.#quic.close();
		} catch {
			// ignore
		}
	}

	async #run(): Promise<void> {
		try {
			await Promise.all([this.#runBidis(), this.#runUnis()]);
		} catch (err) {
			if (!this.#closed) {
				console.error("fatal error running connection", err);
			}
		} finally {
			this.close();
		}
	}

	/**
	 * Publishes a broadcast to the connection.
	 * @param name - The broadcast path to publish
	 * @param broadcast - The broadcast to publish
	 */
	publish(path: Path.Valid, producer: broadcast.Producer) {
		this.#publisher.publish(path, producer);
	}

	/**
	 * Gets an announced reader for the specified prefix.
	 * @param prefix - The prefix for announcements
	 * @returns An Announced instance
	 */
	announced(prefix = Path.empty()): announce.Consumer {
		return this.#subscriber.announced(prefix);
	}

	/**
	 * Consumes a broadcast from the connection.
	 *
	 * @remarks
	 * If the broadcast is not found, a "not found" error will be thrown when requesting any tracks.
	 *
	 * @param broadcast - The path of the broadcast to consume
	 * @returns A Broadcast instance
	 */
	consume(path: Path.Valid): broadcast.Consumer {
		return this.#subscriber.consume(path);
	}

	/**
	 * Watches a broadcast, live only while it is announced.
	 *
	 * @param path - The path of the broadcast to watch
	 * @returns A reactive handle to the broadcast
	 */
	announcedBroadcast(path: Path.Valid): announce.Broadcast {
		return new announce.Broadcast({ connection: this, path });
	}

	/**
	 * Accepts bidi streams (virtual for v14-v16, real for v17) and dispatches.
	 */
	async #runBidis() {
		for (;;) {
			const stream = await this.#session.acceptBi();
			if (!stream) break;

			void this.#runBidi(stream).catch((err: unknown) => {
				console.error("error processing bidi stream", err);
				stream.abort(new Error("bidi stream error"));
			});
		}
	}

	/**
	 * Unified bidi stream dispatch. Reads typeId and routes to handler.
	 * Matches the lite module's runBidi pattern.
	 */
	async #runBidi(stream: Stream) {
		const typeId = await stream.reader.u53();

		switch (typeId) {
			// Draft-18 SUBSCRIBE_NAMESPACE (0x50) and the legacy 0x11 message decode
			// to the same request_id + namespace; the legacy options field is ignored.
			case SubscribeNamespace.id: {
				const msg = await SubscribeNamespace.decode(stream.reader, this.#session.version);
				await this.#publisher.runSubscribeNamespace(msg, stream);
				break;
			}
			case SubscribeNamespaceLegacy.id: {
				const legacy = await SubscribeNamespaceLegacy.decode(stream.reader, this.#session.version);
				const msg = new SubscribeNamespace({ requestId: legacy.requestId, namespace: legacy.namespace });
				await this.#publisher.runSubscribeNamespace(msg, stream);
				break;
			}
			case SubscribeUpdate.id: {
				// REQUEST_UPDATE (0x02) is a follow-up, not a valid initial message
				stream.abort(new Error("unexpected REQUEST_UPDATE as initial message"));
				break;
			}
			// Publisher handles incoming requests
			case Subscribe.id: {
				const msg = await Subscribe.decode(stream.reader, this.#session.version);
				await this.#publisher.runSubscribe(msg, stream);
				break;
			}
			case TrackStatusRequest.id: {
				const msg = await TrackStatusRequest.decode(stream.reader, this.#session.version);
				await this.#publisher.runTrackStatusRequest(msg, stream);
				break;
			}

			// Subscriber handles incoming notifications
			case PublishNamespace.id: {
				const msg = await PublishNamespace.decode(stream.reader, this.#session.version);
				await this.#subscriber.runPublishNamespace(msg, stream);
				break;
			}
			case Publish.id: {
				const msg = await Publish.decode(stream.reader, this.#session.version);
				await this.#subscriber.runPublish(msg, stream);
				break;
			}

			default:
				console.warn(`unexpected bidi stream type: 0x${typeId.toString(16)}`);
				stream.abort(new Error("unexpected stream type"));
		}
	}

	/**
	 * Handles unidirectional streams for media delivery (groups).
	 */
	async #runUnis() {
		const readers = new Readers(this.#quic, this.#session.version);

		for (;;) {
			const stream = await readers.next();
			if (!stream) break;

			this.#runUni(stream)
				.then(() => {
					stream.stop(new Error("cancel"));
				})
				.catch((err: unknown) => {
					console.error("error processing object stream", err);
					stream.stop(err);
				});
		}
	}

	async #runUni(stream: Reader) {
		const header = await Group.decode(stream, this.#session.version);
		await this.#subscriber.handleGroup(header, stream);
	}

	/**
	 * v17+ only: reads GoAway from the setup/control stream.
	 */
	async #runGoAway(controlStream: Stream, version: IetfVersion) {
		try {
			const done = await controlStream.reader.done();
			if (done) return;

			const typeId = await controlStream.reader.u53();
			if (typeId === GoAway.id) {
				const msg = await GoAway.decode(controlStream.reader, version);
				console.warn(`received GOAWAY with redirect URI: ${msg.newSessionUri}`);
			} else {
				console.warn(`unexpected message on setup stream: 0x${typeId.toString(16)}`);
			}
		} catch (err) {
			if (!this.#closed) {
				console.error("error reading setup stream", err);
			}
		} finally {
			this.close();
		}
	}

	/**
	 * Returns a promise that resolves when the connection is closed.
	 * @returns A promise that resolves when closed
	 */
	get closed(): Promise<void> {
		return this.#quic.closed.then(() => undefined);
	}
}
