/**
 * Forgejo/Gitea notifications channel source.
 *
 * Like GitHub, Forgejo has no push transport for notifications, so this source
 * **polls** and emits an event only for *new* threads that match the configured
 * filters. It gates each poll on `GET /notifications/new` — a cheap unread
 * count — and only lists threads when that count is non-zero.
 *
 * The important difference from GitHub: a Forgejo notification thread carries
 * **no `reason` field**. It reports only `{id, unread, updated_at, repository,
 * subject:{type,state,url}}`, so "why am I being told about this" has to be
 * derived by fetching the subject and looking for this account in its
 * `requested_reviewers` or `assignees`. That costs one extra request per *new*
 * thread, never per tick.
 *
 * Filters (env `FORGEJO_NOTIFY_FILTERS`, comma/space list; default
 * `review_requested,assigned_issue,assigned_pr`):
 * - `review_requested` — a pull request review requested from you.
 * - `assigned_issue`   — an issue assigned to you.
 * - `assigned_pr`      — a pull request assigned to you.
 */
import {
	contentDigest,
	permanentIntakeFailure,
	sourceIdentifier,
} from "../task-plane/source-activation.ts";
import type { SourceTaskActivationSink } from "../task-plane/types.ts";
import type { ChannelSource } from "./types.ts";
import { errorMessage, parseList, scopedLog, sinceFrom, trimTrailingSlash } from "./util.ts";

const log = scopedLog("forgejo");

export interface ForgejoConfig {
	/** API base, without the trailing `/api/v1`. */
	url: string;
	token: string;
	/** Login this account answers to; resolved from the API when unset. */
	user?: string | undefined;
	filters: Set<string>;
	pollMs: number;
	/** Mark a matched thread read so the account's inbox does not grow forever. */
	markRead: boolean;
}

const DEFAULT_FILTERS = ["review_requested", "assigned_issue", "assigned_pr"];
const DEFAULT_POLL_MS = 60_000;
const REQUEST_TIMEOUT_MS = 30_000;
const MAX_BACKOFF_MS = 15 * 60_000;

/**
 * The delay before the next poll: the configured interval after a clean tick,
 * doubling per consecutive failure up to `maxMs`. A revoked token or an outage
 * otherwise hammers the forge at full pace forever, one identical log line per
 * tick.
 */
export function backoffDelayMs(pollMs: number, failures: number, maxMs = MAX_BACKOFF_MS): number {
	if (failures <= 0) return pollMs;
	return Math.min(pollMs * 2 ** Math.min(failures, 20), Math.max(pollMs, maxMs));
}

/**
 * Read and discard a response body that will not be parsed. Node's fetch keeps
 * the connection out of the pool until the body is consumed, so skipping this
 * on every error path stalls the pool across a long outage.
 */
function drain(res: Response): void {
	void res.text().catch(() => {});
}

export function forgejoConfigFromEnv(): ForgejoConfig | undefined {
	const token = process.env.FORGEJO_TOKEN;
	// FORGEJO_API_URL exists so a deployment can point the poller at an
	// in-cluster address while FORGEJO_URL stays the public one used in links.
	const url = trimTrailingSlash(process.env.FORGEJO_API_URL || process.env.FORGEJO_URL || "");
	if (!token || !url) return undefined;
	const raw = parseList(process.env.FORGEJO_NOTIFY_FILTERS);
	const filters = new Set(raw.length > 0 ? raw : DEFAULT_FILTERS);
	const pollMs = Number(process.env.FORGEJO_NOTIFY_POLL_MS) || DEFAULT_POLL_MS;
	const markRead = process.env.FORGEJO_NOTIFY_MARK_READ === "1";
	return { url, token, user: process.env.FORGEJO_USER, filters, pollMs, markRead };
}

interface Thread {
	id: number;
	unread?: boolean;
	updated_at: string;
	subject?: { type?: string; url?: string; state?: string };
	repository?: { full_name?: string };
}

interface Subject {
	assignees?: ({ login?: string } | null)[] | null;
	requested_reviewers?: ({ login?: string } | null)[] | null;
}

/** The trusted reasons this source can derive; never free text. */
type Reason = "review_requested" | "assigned_pr" | "assigned_issue";

/** An authenticated, abort-aware request against the forge API. */
type Request_ = (url: string, init?: RequestInit) => Promise<Response>;

interface ForgejoCheckpoint {
	readonly since: string;
	readonly seen: readonly string[];
}

export function createForgejoSource(
	cfg: ForgejoConfig,
	taskSink: SourceTaskActivationSink,
): ChannelSource {
	const api = `${cfg.url}/api/v1`;
	const headers = {
		Authorization: `token ${cfg.token}`,
		Accept: "application/json",
	};

	return {
		async start() {
			const controller = new AbortController();
			// Keys seen in the previous poll only — `since` already excludes
			// anything older, so this just dedups threads sharing the
			// `since`-boundary second. While a tick is failing part-way it also
			// accumulates the threads already emitted, so a retried window never
			// wakes the agent twice for the same thread.
			let seen = new Set<string>();
			// Only notify on threads updated after start-up, anchored to the
			// forge's clock — seeded from the first probe's Date header, never
			// the local one. A local seed ahead of the forge silently filters
			// out every notification in the drift window; behind, the first tick
			// re-emits history.
			let since: string | undefined;
			let login = cfg.user;
			let failures = 0;
			let timer: ReturnType<typeof setTimeout> | undefined;

			const request: Request_ = async (url, init) =>
				await fetch(url, {
					...init,
					headers,
					signal: AbortSignal.any([controller.signal, AbortSignal.timeout(REQUEST_TIMEOUT_MS)]),
				});
			const get = async (url: string): Promise<Response> => await request(url);

			/** Resolve our own login once; without it nothing can be classified. */
			const resolveLogin = async (): Promise<string | undefined> => {
				if (login) return login;
				const res = await get(`${api}/user`);
				if (res.status !== 200) {
					drain(res);
					log(`identity lookup returned HTTP ${res.status}`);
					return undefined;
				}
				const me = (await res.json()) as { login?: string };
				login = me.login;
				if (!login) log("identity lookup returned no login");
				return login;
			};

			/** One poll. Returns false when it should count toward backoff. */
			// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: cursor recovery and acceptance ordering are one atomic poll state machine
			const poll = async (): Promise<boolean> => {
				const me = await resolveLogin();
				if (!me) return false;
				const principal = sourceIdentifier("forgejo", `${cfg.url}\0${me}`);
				if (since === undefined && taskSink.checkpoint) {
					const checkpoint = await taskSink.checkpoint<ForgejoCheckpoint>(principal, "forgejo");
					if (checkpoint) {
						since = checkpoint.since;
						seen = new Set(checkpoint.seen);
					}
				}
				const polled = await pollThreads(api, get, since);
				if (!polled) return false;
				if (since === undefined) {
					// First contact: anchor the cursor to the forge's clock and emit
					// nothing — everything currently listed predates start-up.
					since = polled.since;
					await taskSink.advanceCheckpoint?.(principal, "forgejo", {
						since,
						seen: [],
					} satisfies ForgejoCheckpoint);
					return true;
				}
				const { batch, complete } = await emitNew(
					polled.list,
					seen,
					me,
					cfg,
					api,
					request,
					controller.signal,
					taskSink,
					principal,
				);
				// Advance the cursor only after every thread in the window was
				// classified. A failed subject lookup keeps the old cursor so the
				// thread is retried; the ones already emitted were added to `seen`
				// as they fired, so the retry cannot repeat them.
				if (complete) {
					seen = batch;
					since = polled.since;
					await taskSink.advanceCheckpoint?.(principal, "forgejo", {
						since,
						seen: [...seen],
					} satisfies ForgejoCheckpoint);
				}
				return complete;
			};

			const tick = async (): Promise<void> => {
				try {
					failures = (await poll()) ? 0 : failures + 1;
				} catch (err) {
					if (controller.signal.aborted) return;
					failures += 1;
					log(`poll error: ${errorMessage(err)}`);
				}
			};

			// Self-schedule the next poll only after this one settles, so a slow or
			// hung request can never overlap and race `seen`/`since`.
			const schedule = (): void => {
				timer = setTimeout(
					async () => {
						await tick();
						if (!controller.signal.aborted) schedule();
					},
					backoffDelayMs(cfg.pollMs, failures),
				);
			};
			void (async () => {
				await tick();
				if (!controller.signal.aborted) schedule();
			})();

			return async () => {
				controller.abort();
				if (timer) clearTimeout(timer);
			};
		},
	};
}

/**
 * Fetch the unread threads updated since `since`, gated on a cheap unread count
 * so the list call is skipped entirely when there is nothing new. Returns the
 * threads plus the `since` to use next, or `undefined` when the forge could not
 * be read — in which case `since` must not advance or the missed window is lost.
 *
 * The next cursor always comes from the **probe** response: the probe precedes
 * the listing query, so a thread updated between the two lands after the cursor
 * and is picked up next tick. Taking it from the list response — stamped after
 * the forge evaluated the query — leaves a gap that belongs to neither window.
 * On the very first call `since` is still unset; the caller seeds it from the
 * probe and skips the listing entirely.
 */
async function pollThreads(
	api: string,
	get: (url: string) => Promise<Response>,
	since: string | undefined,
): Promise<{ list: Thread[]; since: string } | undefined> {
	const probe = await get(`${api}/notifications/new`);
	if (probe.status !== 200) {
		drain(probe);
		log(`unread probe returned HTTP ${probe.status}`);
		return undefined;
	}
	const { new: unread } = (await probe.json()) as { new?: number };
	const cursor = sinceFrom(probe);
	if (!unread || since === undefined) return { list: [], since: cursor };

	const res = await get(
		`${api}/notifications?status-types=unread&since=${encodeURIComponent(since)}`,
	);
	if (res.status !== 200) {
		drain(res);
		log(`poll returned HTTP ${res.status}`);
		return undefined;
	}
	return { list: (await res.json()) as Thread[], since: cursor };
}

/**
 * Emit an event for each not-yet-seen thread whose derived reason matches;
 * return the keys seen in this batch (the next poll's dedup set) and whether
 * every thread was classified. A thrown or 5xx subject lookup marks the batch
 * incomplete — that thread is left out of the dedup sets so the caller holds
 * the cursor and retries it, while each thread that *was* processed is added
 * to `seen` immediately so the retried window cannot wake the agent for it
 * again.
 *
 * `summary` is one of our own `Reason` values — deliberately never the issue or
 * pull-request title, which is attacker-controlled text.
 */
// biome-ignore lint/complexity/noExcessiveCognitiveComplexity: every per-item permanent and transient outcome must remain explicit
async function emitNew(
	list: Thread[],
	seen: Set<string>,
	login: string,
	cfg: ForgejoConfig,
	api: string,
	request: Request_,
	signal: AbortSignal,
	taskSink: SourceTaskActivationSink,
	principal: string,
): Promise<{ batch: Set<string>; complete: boolean }> {
	const batch = new Set<string>();
	let complete = true;
	for (const thread of list) {
		const key = `${thread.id}@${thread.updated_at}`;
		if (seen.has(key) || signal.aborted) {
			batch.add(key);
			continue;
		}
		try {
			const reason = await classify(thread, login, api, request);
			if (reason && cfg.filters.has(reason)) {
				const path = subjectPath(thread.subject?.url);
				if (!path) throw new Error("classified Forgejo subject has no exact path");
				const acceptance = await taskSink.accept({
					principal,
					source: "forgejo",
					providerEventId: sourceIdentifier("event", key),
					providerDedupeKey: sourceIdentifier("event", key),
					nativeLocator: {
						threadId: String(thread.id),
						revision: thread.updated_at,
						subjectPath: path,
						repository: thread.repository?.full_name ?? "",
						reason,
					},
					receivedAt: thread.updated_at,
					conversationKey: sourceIdentifier("conversation", path),
					parts: [
						{
							data: {
								threadId: thread.id,
								revision: thread.updated_at,
								path,
								repository: thread.repository?.full_name ?? "",
								reason,
							},
						},
					],
					contentDigest: contentDigest({ thread, reason }),
				});
				if (cfg.markRead) {
					await deliverMarkRead(taskSink, acceptance.taskId, thread, api, request);
				}
			} else {
				await taskSink.recordEvidence?.({
					evidenceId: sourceIdentifier("evidence", `${principal}\0${key}`),
					source: "forgejo",
					kind: "permanent-non-work",
					detail: { threadId: String(thread.id), revision: thread.updated_at },
				});
			}
		} catch (err) {
			if (signal.aborted) return { batch, complete: false };
			const permanent = permanentIntakeFailure(err);
			if (permanent) {
				await taskSink.recordEvidence?.({
					evidenceId: sourceIdentifier("evidence", `${principal}\0${key}`),
					source: "forgejo",
					kind: permanent.kind,
					detail: { threadId: String(thread.id), revision: thread.updated_at },
				});
				batch.add(key);
				seen.add(key);
				continue;
			}
			complete = false;
			log(`subject lookup failed for thread ${thread.id}: ${errorMessage(err)}`);
			continue;
		}
		batch.add(key);
		seen.add(key);
	}
	return { batch, complete };
}

/**
 * Derive why this account was notified. Forgejo threads carry no `reason`, so
 * the subject is fetched and inspected; an unrecognized subject yields no
 * reason rather than a guess, and a transient lookup failure throws so the
 * thread is retried rather than silently lost.
 */
async function classify(
	thread: Thread,
	login: string,
	api: string,
	request: Request_,
): Promise<Reason | undefined> {
	const type = thread.subject?.type; // "Pull" | "Issue" | "Commit" | "Repository"
	if (type !== "Pull" && type !== "Issue") return undefined;
	// Re-root the subject onto the configured API base rather than fetching the
	// URL the payload carries. A notification's `subject.url` is absolute and
	// points at the forge's public host; a deployment that reaches the forge on
	// an internal address cannot necessarily reach that one, and the request
	// fails in a way that yields no reason and therefore no wake — every
	// notification silently lost. Only the path is taken from the payload.
	const path = subjectPath(thread.subject?.url);
	if (!path) return undefined;

	const res = await request(`${api}${path}`);
	if (res.status !== 200) {
		drain(res);
		// A server-side failure is transient: throw so the caller retries the
		// thread instead of dropping the notification. Anything else (404 for a
		// deleted subject, 403) will not improve on retry — drop it.
		if (res.status >= 500 || res.status === 429) {
			throw new Error(`subject lookup returned HTTP ${res.status}`);
		}
		log(`subject lookup returned HTTP ${res.status}`);
		return undefined;
	}
	const subject = (await res.json()) as Subject;

	if (type === "Pull" && includesLogin(subject.requested_reviewers, login)) {
		return "review_requested";
	}
	if (includesLogin(subject.assignees, login)) {
		return type === "Pull" ? "assigned_pr" : "assigned_issue";
	}
	return undefined;
}

/**
 * The API path of a notification subject, e.g. `/repos/o/r/issues/3`. Returns
 * undefined when the payload carries no usable path, so an unparseable or
 * hostile value is dropped rather than fetched.
 */
export function subjectPath(raw: string | undefined): string | undefined {
	if (!raw) return undefined;
	let pathname: string;
	try {
		pathname = new URL(raw).pathname;
	} catch {
		return undefined;
	}
	const marker = "/api/v1/";
	const at = pathname.indexOf(marker);
	if (at === -1) return undefined;
	return pathname.slice(at + marker.length - 1);
}

function includesLogin(
	users: ({ login?: string } | null)[] | null | undefined,
	login: string,
): boolean {
	return (users ?? []).some((user) => user?.login === login);
}

/** Exact idempotent acknowledgment, invoked only after Task acceptance. */
async function markRead(thread: Thread, api: string, request: Request_): Promise<void> {
	// `id` is typed as a number but arrives as unvalidated JSON; a string value
	// would path-traverse out of this endpoint.
	if (!Number.isSafeInteger(thread.id)) {
		log(`ignoring a notification with a non-numeric id`);
		return;
	}
	const res = await request(`${api}/notifications/threads/${thread.id}?to-status=read`, {
		method: "PATCH",
	});
	drain(res);
	if (res.status === 404) return;
	if (res.status >= 400) {
		throw Object.assign(new Error(`mark-read returned HTTP ${res.status}`), {
			status: res.status,
		});
	}
}

async function deliverMarkRead(
	taskSink: SourceTaskActivationSink,
	taskId: string,
	thread: Thread,
	api: string,
	request: Request_,
): Promise<void> {
	if (!taskSink.deliver) throw new Error("Forgejo task delivery is not configured");
	await taskSink.deliver(
		{
			taskId,
			source: "forgejo",
			operationId: `mark-read:${thread.id}`,
			payloadDigest: contentDigest({ status: "read" }),
			recovery: "idempotent",
		},
		async () => {
			await markRead(thread, api, request);
			return String(thread.id);
		},
	);
}