import { getMailbox, insertOpsEvent, insertTelegramLink } from "../db/d1";
import {
	claimTelegramTopic,
	forgetTelegramTopic,
	getTelegramTopicMapping,
} from "../db/telegram-topics";
import {
	createForumTopic,
	isMissingTopicError,
	readTelegramConfig,
	sendMessage,
	TelegramApiError,
	type TelegramConfig,
	type TelegramMessage,
} from "./api";
import { chatSupportsTopics, resolveTelegramChatId } from "./binding";
import { inboundNotificationKeyboard } from "./cards";
import { renderInboundNotification } from "./format";
import { isFirstContact } from "./index-rows";
import { fetchMailboxAttachments } from "./messages";
import { flushTelegramDigest, isQuietNow, isSenderMuted, retainNotification } from "./noise";

export type InboundNotificationInput = {
	mailboxId: string;
	mailboxAddress: string;
	messageLocalId: string;
	threadId: string;
	subject: string | null;
	fromAddr: string;
	snippet: string | null;
	hasAttachments: boolean;
};

/**
 * What a mailbox is called in the Telegram sidebar.
 *
 * The operator's own words for his own mailbox: the name he gave the topic
 * itself when he gave one (telegram_topics.topic_name, see migration 0020),
 * else display_name, else the address. Pointedly NOT the email subject: a topic
 * name is permanent chrome in the operator's client, and deriving it from
 * inbound mail would hand every stranger who can send to this address a pen for
 * the sidebar.
 *
 * The stored name comes first because display_name is also the From name of
 * every reply the mailbox sends: tying the two together meant a topic could not
 * be renamed without changing how the mailbox signs its mail.
 */
export function topicNameFor(
	mailbox: { display_name: string | null; primary_address: string } | null,
	fallbackAddress: string,
	storedName?: string | null,
): string {
	return (
		storedName?.trim() ||
		mailbox?.display_name?.trim() ||
		mailbox?.primary_address ||
		fallbackAddress
	);
}

/**
 * Posts into a mailbox's topic, or into the chat itself when there is none.
 *
 * The one send every mailbox-addressed message goes through -- real cards and
 * the operator's delivery test alike -- so the test exercises the exact
 * message_thread_id handling a card gets, not a copy of it. A falsy topic id is
 * dropped rather than sent as 0: Telegram then files the message under General,
 * which is also what it silently does with a thread id it does not recognise.
 * That silence is why callers that care compare the returned message_thread_id
 * with the one they asked for.
 */
export async function sendToMailboxTopic(
	config: TelegramConfig,
	input: {
		chatId: string;
		topicId: number | null;
		text: string;
		replyMarkup?: Record<string, unknown>;
	},
): Promise<TelegramMessage> {
	return sendMessage(config, {
		chatId: input.chatId,
		text: input.text,
		messageThreadId: input.topicId,
		replyMarkup: input.replyMarkup,
	});
}

/**
 * Retryable in the sense the notification queue means it: worth redelivering.
 *
 * Mirrors the consumer's own rule (429 and 5xx are about the moment, other 4xx
 * are about the request) so a topic that cannot be created *right now* costs a
 * redelivery instead of silently demoting this mailbox to flat messages, while a
 * bot that simply lacks manage_topics rights degrades once and stays useful.
 */
function isRetryable(error: unknown): boolean {
	if (error instanceof TelegramApiError) {
		return error.statusCode === 429 || error.statusCode >= 500;
	}
	return true;
}

/**
 * The topic this mailbox's mail belongs in, creating it the first time.
 *
 * One topic per mailbox, not per email thread: mailboxes are a bounded namespace
 * the operator curated, threads are unbounded input from strangers. See
 * migrations/d1/0010_telegram_topics.sql.
 *
 * Returns null when the chat has no topics at all, which is the ordinary case: a
 * private chat with the bot is a flat message stream, and replies there thread
 * through reply_to_message exactly as they always have.
 */
async function resolveTopicId(
	config: TelegramConfig,
	env: Env,
	input: InboundNotificationInput,
	chatId: string,
	/**
	 * The operator-chosen name of a topic that was just found deleted, so the one
	 * recreated in its place keeps it. Null/absent falls back to the mailbox.
	 */
	storedName: string | null = null,
): Promise<number | null> {
	if (!(await chatSupportsTopics(env))) {
		return null;
	}
	const existing = await getTelegramTopicMapping(env.INDEX_DB, chatId, input.mailboxId);
	if (existing !== null) {
		return existing.topic_id;
	}
	const mailbox = await getMailbox(env.INDEX_DB, input.mailboxId);
	try {
		const topic = await createForumTopic(config, {
			chatId,
			name: topicNameFor(mailbox, input.mailboxAddress, storedName),
		});
		return claimTelegramTopic(env.INDEX_DB, {
			chatId,
			mailboxId: input.mailboxId,
			topicId: topic.message_thread_id,
			// Carried over, not invented: an auto-created topic stays NULL so it keeps
			// following the mailbox, a recreated one keeps the name it was given.
			topicName: storedName,
		});
	} catch (error) {
		if (isRetryable(error)) {
			throw error;
		}
		await insertOpsEvent(env.INDEX_DB, {
			id: crypto.randomUUID(),
			event_type: "telegram.topic_unavailable",
			severity: "warning",
			subject: input.mailboxId,
			payload_json: JSON.stringify({
				error: error instanceof Error ? error.message : String(error),
				hint: "Falling back to plain messages; replies still work via reply_to_message.",
			}),
		});
		return null;
	}
}

/**
 * Why a notification did not go out when nothing actually failed. These are
 * decisions about this deployment, identical on every redelivery, so the
 * consumer acks them instead of retrying a bridge that is simply off.
 */
export type NotifyOutcome =
	| { status: "sent" }
	| {
			status: "skipped";
			reason: "bridge_disabled" | "no_chat" | "muted_sender" | "quiet_hours";
	  };

/**
 * Pushes a new-mail card to Telegram and records the link that makes a reply to
 * it resolvable.
 *
 * Throws on transport failure, deliberately: this runs on the notification
 * queue, whose entire reason to exist is that Cloudflare can redeliver what
 * Telegram refused. The older swallow-everything contract belonged to the days
 * when this was awaited on the ingest ack path, where a throw would have
 * re-delivered an already-stored email.
 */
export async function deliverInboundNotification(
	env: Env,
	input: InboundNotificationInput,
): Promise<NotifyOutcome> {
	const config = readTelegramConfig(env);
	if (!config) {
		return { status: "skipped", reason: "bridge_disabled" };
	}
	// No operator check here on purpose: the allowlist governs who may give the bot
	// orders, not who may be told about mail. An unpaired bridge simply has no
	// adopted chat yet, and that is the skip that fires.
	// The adopted chat wins over anything a deploy manifest claims: it is the chat
	// the operator actually spoke from, observed rather than transcribed.
	const chatId = await resolveTelegramChatId(env);
	if (!chatId) {
		return { status: "skipped", reason: "no_chat" };
	}

	// Checked here rather than at ingest, and only ever against notifying: a muted
	// sender's mail is stored, indexed, searchable and answerable exactly like any
	// other. The decision is "do not interrupt me", not "do not keep this".
	if (await isSenderMuted(env.INDEX_DB, input.fromAddr)) {
		return { status: "skipped", reason: "muted_sender" };
	}

	// Before the quiet-hours check, so the first email after the window closes is
	// also what releases everything the window held. The hourly cron does the same
	// for a morning with no new mail; see docs/plans/telegram-roadmap.md.
	await flushTelegramDigest(env).catch((error: unknown) => {
		console.warn("telegram.digest_flush_failed", {
			error: error instanceof Error ? error.message : String(error),
		});
	});
	if (await isQuietNow(env.INDEX_DB)) {
		await retainNotification(env.INDEX_DB, {
			chatId,
			mailboxId: input.mailboxId,
			mailboxAddress: input.mailboxAddress,
			messageLocalId: input.messageLocalId,
			threadId: input.threadId,
			fromAddr: input.fromAddr,
		});
		return { status: "skipped", reason: "quiet_hours" };
	}

	// Two extra reads, both earned. The badge is one indexed lookup and is the
	// single most useful thing a card can say about an address that receives cold
	// mail; the filenames are one Durable Object round trip taken only for the
	// minority of mail that carries files, and a failure degrades the card to the
	// old boolean rather than costing the notification.
	const [firstContact, attachments] = await Promise.all([
		isFirstContact(env.INDEX_DB, {
			fromAddr: input.fromAddr,
			messageLocalId: input.messageLocalId,
		}),
		input.hasAttachments
			? fetchMailboxAttachments(env, input.mailboxId, input.messageLocalId)
			: Promise.resolve(null),
	]);

	const text = renderInboundNotification({
		fromAddr: input.fromAddr,
		mailboxAddress: input.mailboxAddress,
		subject: input.subject,
		snippet: input.snippet,
		hasAttachments: input.hasAttachments,
		attachments: attachments?.map((attachment) => ({
			filename: attachment.filename,
			size: attachment.size,
		})),
		firstContact,
	});
	// The card carries what can be done with the mail, not just what it says: most
	// email is dispatched rather than answered, and a card with no buttons made
	// Telegram useful only for the minority that needs a sentence back.
	const replyMarkup = await inboundNotificationKeyboard(env, {
		mailboxId: input.mailboxId,
		threadId: input.threadId,
		fromAddr: input.fromAddr,
		hasAttachments: input.hasAttachments,
	});

	let topicId = await resolveTopicId(config, env, input, chatId);
	let sent: TelegramMessage;
	try {
		sent = await sendToMailboxTopic(config, { chatId, topicId, text, replyMarkup });
	} catch (error) {
		// The operator deleted the topic. Nothing here is broken -- deleting a topic
		// is a normal thing to do -- but the stored id is now a dead address, and
		// without healing it every future email for this mailbox would fail the same
		// way, which is exactly the silent failure this bridge keeps rediscovering.
		if (topicId === null || !isMissingTopicError(error)) {
			throw error;
		}
		// Read before forgetting: a name the operator chose must outlive the topic it
		// was first given to, or deleting a topic would silently rename the mailbox's
		// next one back to its display_name.
		const stale = await getTelegramTopicMapping(env.INDEX_DB, chatId, input.mailboxId);
		await forgetTelegramTopic(env.INDEX_DB, chatId, input.mailboxId);
		await insertOpsEvent(env.INDEX_DB, {
			id: crypto.randomUUID(),
			event_type: "telegram.topic_recreated",
			severity: "info",
			subject: input.mailboxId,
			payload_json: JSON.stringify({ chatId, staleTopicId: topicId }),
		}).catch(() => undefined);
		// Exactly one retry: a second miss means the failure is not the topic, and
		// looping on it would burn the chat's rate-limit budget instead of the
		// queue's backoff.
		topicId = await resolveTopicId(config, env, input, chatId, stale?.topic_name ?? null);
		sent = await sendToMailboxTopic(config, { chatId, topicId, text, replyMarkup });
	}
	// Inside the retried path on purpose: without this row a reply typed in
	// Telegram resolves to no thread and is dropped, and a card the operator sees
	// twice is a cheaper failure than an answer that never leaves the phone.
	await insertTelegramLink(env.INDEX_DB, {
		chat_id: chatId,
		message_id: sent.message_id,
		mailbox_id: input.mailboxId,
		thread_id: input.threadId,
		message_local_id: input.messageLocalId,
		topic_id: topicId,
	});
	return { status: "sent" };
}