import { WebSocket } from "undici";

import { config } from "./config";
import { Frames } from "./frames";
import { dispatcher, get, partsOf } from "./net";
import type { Held } from "./tabs";

type Action = "create" | "update" | "delete";

interface Seen {
	id: number;
	action: Action;
	cid?: string;
	live: boolean;
	at: number;
}

// /stats lags the newest id, so start the replay a bit before it
const REPLAY = 1_000;
// frames go out when the db write commits, so this only covers the db hop
const BUDGET = 10_000;

const uriOf = (record: { did: string; collection: string; rkey: string }) =>
	`at://${record.did}/${record.collection}/${record.rkey}`;

// opened before the first flow so a check can look back at earlier frames
class Watcher extends Frames<Seen> {
	private socket?: WebSocket;
	private floor = -1;
	private last = -1;
	private stopped = true;
	private retries = 0;
	private downSince?: number;
	private lastWords = "";

	get running() {
		return !this.stopped;
	}

	async start() {
		if (this.running) return;
		const { status, body } = await get(`${config.hydrantUrl}/stats`);
		const stored = body?.counts?.events;
		if (status !== 200 || typeof stored !== "number") {
			throw new Error(
				`hydrant /stats should count its events but answered ${status} ${JSON.stringify(body)}`
			);
		}
		this.floor = Math.max(0, stored - REPLAY);
		this.stopped = false;
		await this.connect().catch((cause) => {
			this.stop();
			throw cause;
		});
	}

	stop() {
		this.stopped = true;
		this.socket?.close();
	}

	// take it before the write goes through, the frames it causes all get higher ids
	mark() {
		return Math.max(this.last, this.floor - 1);
	}

	// a dead socket must never look like hydrant being quiet
	describe(since: number) {
		const parts = [`the watcher is at #${this.last}`];
		if (since < this.forgotUpTo) parts.push(`it forgot frames up to #${this.forgotUpTo}`);
		if (this.downSince !== undefined) {
			const down = ((Date.now() - this.downSince) / 1000).toFixed(1);
			parts.push(
				`it has been disconnected for ${down}s (${this.lastWords || "no reason given"})`
			);
		}
		return parts.join(", ");
	}

	// a reconnect resumes at the next id and hydrant replays the record frames missed while down
	private connect() {
		const cursor = this.last >= 0 ? this.last + 1 : this.floor;
		const url = `${config.hydrantUrl.replace(/^http/, "ws")}/stream?cursor=${cursor}`;
		const socket = new WebSocket(url, { dispatcher });
		this.socket = socket;
		socket.addEventListener("message", (event) => this.take(JSON.parse(String(event.data))));
		socket.addEventListener("open", () => {
			this.downSince = undefined;
			this.retries = 0;
		});
		socket.addEventListener("close", (event) => {
			if (this.stopped) return;
			this.downSince ??= Date.now();
			this.lastWords ||= `close ${event.code} ${event.reason}`.trim();
			const wait = Math.min(250 * 2 ** this.retries++, 2_000);
			setTimeout(() => {
				if (!this.stopped) this.connect().catch(() => {});
			}, wait);
		});
		return new Promise<void>((opened, failed) => {
			socket.addEventListener("open", () => opened(), { once: true });
			socket.addEventListener("error", () => failed(new Error(`${url} would not open`)), {
				once: true
			});
		});
	}

	private take(frame: any) {
		if (frame.type === "error") {
			this.lastWords = `hydrant said ${frame.error}: ${frame.message}`;
			return;
		}
		if (typeof frame.id !== "number" || frame.id <= this.last) return;
		this.last = frame.id;
		this.lastWords = "";
		if (frame.type === "record") this.keep(frame.id, frame.record);
		this.wake();
	}

	private keep(id: number, record: any) {
		const uri = uriOf(record);
		this.append(uri, {
			id,
			action: record.action,
			cid: record.cid,
			live: record.live,
			at: Date.now()
		});
	}
}

export const watcher = new Watcher();

const waitFor = async (what: string, check: () => [done: boolean, saw: string, newest?: Seen]) => {
	const deadline = Date.now() + BUDGET;
	for (;;) {
		const [done, saw, newest] = check();
		if (done) return newest;
		const left = deadline - Date.now();
		if (left <= 0) throw new Error(`${what}, gave up after ${BUDGET / 1000}s: ${saw}`);
		await watcher.changed(left);
	}
};

const told = (seen: Seen) =>
	`${seen.action}${seen.cid ? ` ${seen.cid}` : ""} #${seen.id}${seen.live ? "" : " (backfill)"}`;

const framesOf = (uri: string, since: number) => {
	if (!watcher.running) {
		throw new Error(
			`hydrant's stream should be watched for ${uri} but this process never opened it`
		);
	}
	const frames = watcher.seen(uri).filter((seen) => seen.id > since);
	const after = since < 0 ? "since the watcher opened" : `after #${since}`;
	const said = frames.length
		? `${after} it announced ${frames.slice(-5).map(told).join(", ")}`
		: `it announced nothing for that uri ${after}`;
	return { frames, said: `${said}, ${watcher.describe(since)}` };
};

const replacedBy = (uri: string, seen: Seen) => {
	const before = watcher
		.seen(uri)
		.filter((other) => other.id < seen.id && other.cid !== seen.cid)
		.at(-1);
	return before?.action === "delete" ? undefined : before;
};

export const checkStream = (
	{ uri, cid }: { uri: string; cid?: string },
	how: Held["method"] | "settled",
	since = -1
) => {
	if (cid === undefined)
		return waitFor(`hydrant's stream should announce the delete of ${uri}`, () => {
			const { frames, said } = framesOf(uri, since);
			const newest = frames.at(-1);
			return [newest?.action === "delete", said, newest];
		});
	const wanted = how === "create" ? "a create" : "a create or update";
	return waitFor(`hydrant's stream should announce ${wanted} of ${uri} at ${cid}`, () => {
		const { frames, said } = framesOf(uri, since);
		const newest = frames.at(-1);
		// only the newest frame counts, so an older cid coming back after it fails too
		if (newest?.cid !== cid || newest.action === "delete") return [false, said];
		const ours = frames.filter((seen) => seen.cid === cid);
		if (how === "create") return [ours.some((seen) => seen.action === "create"), said, newest];
		// a backfill says create for old records too, so only a live create counts as new
		const stillThere = ours.map((seen) =>
			seen.live && seen.action === "create" ? replacedBy(uri, seen) : undefined
		);
		if (stillThere.every(Boolean))
			return [
				false,
				`${said}, but ${told(stillThere[0]!)} was still there so it had to be an update`
			];
		return [true, said, newest];
	});
};

// hydrant only indexes org.tangled.*, so other collections never reach the stream
export const streamCarries = (uri: string) => partsOf(uri).collection.startsWith("org.tangled.");