from typing import TYPE_CHECKING

import sentry_sdk
from sentry_sdk.crons import MonitorStatus, capture_checkin
from sentry_sdk.integrations import DidNotEnable
from sentry_sdk.integrations.celery.utils import (
    _get_humanized_interval,
    _now_seconds_since_epoch,
)
from sentry_sdk.utils import (
    logger,
    match_regex_list,
)

if TYPE_CHECKING:
    from collections.abc import Callable
    from typing import Any, Optional, TypeVar, Union

    from sentry_sdk._types import (
        MonitorConfig,
        MonitorConfigScheduleType,
        MonitorConfigScheduleUnit,
    )

    F = TypeVar("F", bound=Callable[..., Any])


try:
    from celery import Celery, Task  # type: ignore
    from celery.beat import Scheduler  # type: ignore
    from celery.schedules import crontab, schedule  # type: ignore
    from celery.signals import (  # type: ignore
        task_failure,
        task_retry,
        task_success,
    )
except ImportError:
    raise DidNotEnable("Celery not installed")

try:
    from redbeat.schedulers import RedBeatScheduler  # type: ignore
except ImportError:
    RedBeatScheduler = None


def _get_headers(task: "Task") -> "dict[str, Any]":
    headers = task.request.get("headers") or {}

    # flatten nested headers
    if "headers" in headers:
        headers.update(headers["headers"])
        del headers["headers"]

    headers.update(task.request.get("properties") or {})

    return headers


def _get_monitor_config(
    celery_schedule: "Any", app: "Celery", monitor_name: str
) -> "MonitorConfig":
    monitor_config: "MonitorConfig" = {}
    schedule_type: "Optional[MonitorConfigScheduleType]" = None
    schedule_value: "Optional[Union[str, int]]" = None
    schedule_unit: "Optional[MonitorConfigScheduleUnit]" = None

    if isinstance(celery_schedule, crontab):
        schedule_type = "crontab"
        schedule_value = (
            "{0._orig_minute} "
            "{0._orig_hour} "
            "{0._orig_day_of_month} "
            "{0._orig_month_of_year} "
            "{0._orig_day_of_week}".format(celery_schedule)
        )
    elif isinstance(celery_schedule, schedule):
        schedule_type = "interval"
        (schedule_value, schedule_unit) = _get_humanized_interval(
            celery_schedule.seconds
        )

        if schedule_unit == "second":
            logger.warning(
                "Intervals shorter than one minute are not supported by Sentry Crons. Monitor '%s' has an interval of %s seconds. Use the `exclude_beat_tasks` option in the celery integration to exclude it.",
                monitor_name,
                schedule_value,
            )
            return {}

    else:
        logger.warning(
            "Celery schedule type '%s' not supported by Sentry Crons.",
            type(celery_schedule),
        )
        return {}

    monitor_config["schedule"] = {}
    monitor_config["schedule"]["type"] = schedule_type
    monitor_config["schedule"]["value"] = schedule_value

    if schedule_unit is not None:
        monitor_config["schedule"]["unit"] = schedule_unit

    monitor_config["timezone"] = (
        (
            hasattr(celery_schedule, "tz")
            and celery_schedule.tz is not None
            and str(celery_schedule.tz)
        )
        or app.timezone
        or "UTC"
    )

    return monitor_config


def _apply_crons_data_to_schedule_entry(
    scheduler: "Any",
    schedule_entry: "Any",
    integration: "sentry_sdk.integrations.celery.CeleryIntegration",
) -> None:
    """
    Add Sentry Crons information to the schedule_entry headers.
    """
    if not integration.monitor_beat_tasks:
        return

    monitor_name = schedule_entry.name

    task_should_be_excluded = match_regex_list(
        monitor_name, integration.exclude_beat_tasks
    )
    if task_should_be_excluded:
        return

    celery_schedule = schedule_entry.schedule
    app = scheduler.app

    monitor_config = _get_monitor_config(celery_schedule, app, monitor_name)

    is_supported_schedule = bool(monitor_config)
    if not is_supported_schedule:
        return

    headers = schedule_entry.options.pop("headers", {})
    headers.update(
        {
            "sentry-monitor-slug": monitor_name,
            "sentry-monitor-config": monitor_config,
        }
    )

    check_in_id = capture_checkin(
        monitor_slug=monitor_name,
        monitor_config=monitor_config,
        status=MonitorStatus.IN_PROGRESS,
    )
    headers.update({"sentry-monitor-check-in-id": check_in_id})

    # Set the Sentry configuration in the options of the ScheduleEntry.
    # Those will be picked up in `apply_async` and added to the headers.
    schedule_entry.options["headers"] = headers


def _wrap_beat_scheduler(
    original_function: "Callable[..., Any]",
) -> "Callable[..., Any]":
    """
    Makes sure that:
    - a new Sentry trace is started for each task started by Celery Beat and
      it is propagated to the task.
    - the Sentry Crons information is set in the Celery Beat task's
      headers so that is monitored with Sentry Crons.

    After the patched function is called,
    Celery Beat will call apply_async to put the task in the queue.
    """
    # Patch only once
    # Can't use __name__ here, because some of our tests mock original_apply_entry
    already_patched = "sentry_patched_scheduler" in str(original_function)
    if already_patched:
        return original_function

    from sentry_sdk.integrations.celery import CeleryIntegration

    def sentry_patched_scheduler(*args: "Any", **kwargs: "Any") -> None:
        integration = sentry_sdk.get_client().get_integration(CeleryIntegration)
        if integration is None:
            return original_function(*args, **kwargs)

        # Tasks started by Celery Beat start a new Trace
        scope = sentry_sdk.get_isolation_scope()
        scope.set_new_propagation_context()
        scope._name = "celery-beat"

        scheduler, schedule_entry = args
        _apply_crons_data_to_schedule_entry(scheduler, schedule_entry, integration)

        return original_function(*args, **kwargs)

    return sentry_patched_scheduler


def _patch_beat_apply_entry() -> None:
    Scheduler.apply_entry = _wrap_beat_scheduler(Scheduler.apply_entry)


def _patch_redbeat_apply_async() -> None:
    if RedBeatScheduler is None:
        return

    RedBeatScheduler.apply_async = _wrap_beat_scheduler(RedBeatScheduler.apply_async)


def _setup_celery_beat_signals(monitor_beat_tasks: bool) -> None:
    if monitor_beat_tasks:
        task_success.connect(crons_task_success)
        task_failure.connect(crons_task_failure)
        task_retry.connect(crons_task_retry)


def crons_task_success(sender: "Task", **kwargs: "dict[Any, Any]") -> None:
    logger.debug("celery_task_success %s", sender)
    headers = _get_headers(sender)

    if "sentry-monitor-slug" not in headers:
        return

    monitor_config = headers.get("sentry-monitor-config", {})

    start_timestamp_s = headers.get("sentry-monitor-start-timestamp-s")

    capture_checkin(
        monitor_slug=headers["sentry-monitor-slug"],
        monitor_config=monitor_config,
        check_in_id=headers["sentry-monitor-check-in-id"],
        duration=(
            _now_seconds_since_epoch() - float(start_timestamp_s)
            if start_timestamp_s
            else None
        ),
        status=MonitorStatus.OK,
    )


def crons_task_failure(sender: "Task", **kwargs: "dict[Any, Any]") -> None:
    logger.debug("celery_task_failure %s", sender)
    headers = _get_headers(sender)

    if "sentry-monitor-slug" not in headers:
        return

    monitor_config = headers.get("sentry-monitor-config", {})

    start_timestamp_s = headers.get("sentry-monitor-start-timestamp-s")

    capture_checkin(
        monitor_slug=headers["sentry-monitor-slug"],
        monitor_config=monitor_config,
        check_in_id=headers["sentry-monitor-check-in-id"],
        duration=(
            _now_seconds_since_epoch() - float(start_timestamp_s)
            if start_timestamp_s
            else None
        ),
        status=MonitorStatus.ERROR,
    )


def crons_task_retry(sender: "Task", **kwargs: "dict[Any, Any]") -> None:
    logger.debug("celery_task_retry %s", sender)
    headers = _get_headers(sender)

    if "sentry-monitor-slug" not in headers:
        return

    monitor_config = headers.get("sentry-monitor-config", {})

    start_timestamp_s = headers.get("sentry-monitor-start-timestamp-s")

    capture_checkin(
        monitor_slug=headers["sentry-monitor-slug"],
        monitor_config=monitor_config,
        check_in_id=headers["sentry-monitor-check-in-id"],
        duration=(
            _now_seconds_since_epoch() - float(start_timestamp_s)
            if start_timestamp_s
            else None
        ),
        status=MonitorStatus.ERROR,
    )