"""Graceful shutdown helpers shared by signal handlers and the settings UI."""

from __future__ import annotations

import os
import signal
import sys
import threading
from collections.abc import Callable

from fastapi import FastAPI


def begin_shutdown(app: FastAPI) -> None:
	"""Wake long-poll waiters; safe to call multiple times."""
	app.state.shutting_down = True
	shutdown_event = getattr(app.state, "shutdown_event", None)
	if shutdown_event is not None:
		shutdown_event.set()


def request_graceful_shutdown(app: FastAPI) -> None:
	"""Begin graceful shutdown and signal the server process when supported."""
	begin_shutdown(app)
	if getattr(app.state, "can_signal_shutdown", False):
		os.kill(os.getpid(), signal.SIGTERM)


def install_graceful_shutdown_wakeup(app: FastAPI) -> bool:
	"""Chain OS signals so long-poll waiters wake before uvicorn shuts down.

	Returns whether SIGTERM may be sent to this process later (e.g. from the
	settings shutdown button). Skipped off the main thread.
	"""
	if threading.current_thread() is not threading.main_thread():
		app.state.can_signal_shutdown = False
		return False
	signals: tuple[int, ...] = (signal.SIGINT, signal.SIGTERM)
	if sys.platform == "win32":
		signals = signals + (signal.SIGBREAK,)
	installed = False
	for sig in signals:

		def make_chain(
			previous: Callable[[int, object | None], object] | int | None,
		) -> Callable[[int, object | None], None]:
			def handler(signum: int, frame: object | None) -> None:
				begin_shutdown(app)
				if callable(previous) and previous not in (signal.SIG_DFL, signal.SIG_IGN):
					previous(signum, frame)

			return handler

		try:
			prev = signal.getsignal(sig)
			signal.signal(sig, make_chain(prev))
			installed = True
		except ValueError:
			continue
	app.state.can_signal_shutdown = installed
	return installed