# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Redis backend for prometheus values."""

import datetime as dt
import os
from threading import Event, Thread
from typing import Any, override
from urllib.parse import urlsplit

from django.conf import settings
from django.test.utils import TestContextDecorator, override_settings
from redis import Redis

# For testing, a pool of otherwise anonymous FakeRedis instances are made
# available by ID
_fake_redis_pool: dict[int, Redis] = {}


def redis_client() -> Redis:
    """
    Create a redis client for PROMETHEUS_REDIS_URL.

    Configure the redis database via a URL in PROMETHEUS_REDIS_URL of the form
    redis://localhost:6379/0
    """
    url = getattr(settings, "PROMETHEUS_REDIS_URL")
    assert url
    parsed_url = urlsplit(url)
    assert parsed_url.path.startswith("/")
    assert parsed_url.path[1:].isdigit()
    port = parsed_url.port or 6379
    db = int(parsed_url.path[1:])

    if parsed_url.scheme == "fakeredis":
        from fakeredis import FakeRedis

        if db not in _fake_redis_pool:
            _fake_redis_pool[db] = FakeRedis()
        return _fake_redis_pool[db]

    assert parsed_url.scheme == "redis"
    assert parsed_url.hostname
    return Redis(host=parsed_url.hostname, port=port, db=db)


# For each process identifier, a list of keys that should be kept from expiring
_live_metrics: dict[str, set[str]] = {}


def _key_expiry() -> dt.timedelta:
    """Return the configured expiry for multiprocess keys."""
    return dt.timedelta(
        seconds=int(os.environ.get("PROMETHEUS_REDIS_REFRESH_TTL", 20))
    )


class KeepMetricsAliveThread(Thread):
    """A daemon thread that keeps metrics from expiring as long as we live."""

    stop: Event
    identifier: str
    client: Redis

    @override
    def __init__(
        self, identifier: str, client: Redis, *args: Any, **kwargs: Any
    ) -> None:
        self.stop = Event()
        self.identifier = identifier
        self.client = client
        super().__init__(*args, **kwargs)

    def loop_wait(self, delay: float) -> bool:
        """Patchable wait step."""
        return self.stop.wait(delay)

    @override
    def run(self) -> None:
        delay = float(os.environ.get("PROMETHEUS_REDIS_REFRESH_FREQUENCY", 10))
        expiry = _key_expiry()
        while not self.loop_wait(delay):
            for key in _live_metrics[self.identifier]:
                self.client.expire(key, expiry)


_daemon_threads: dict[str, KeepMetricsAliveThread] = {}


def _keep_key_from_expiring(identifier: str, key: str) -> None:
    """Stop key for process identifier from expiring as long as we are alive."""
    _live_metrics.setdefault(identifier, set()).add(key)
    if identifier not in _daemon_threads:
        thread = KeepMetricsAliveThread(
            identifier=identifier, client=redis_client(), daemon=True
        )
        thread.start()
        _daemon_threads[identifier] = thread


def mark_process_dead(identifier: str | int) -> None:
    """Immediately expire all live* metrics for process identifier."""
    thread = _daemon_threads.pop(str(identifier), None)
    if thread is not None:
        thread.stop.set()
        thread.join()

    keys = _live_metrics.pop(str(identifier), None)
    if not keys:
        return
    redis_client().delete(*keys)


class private_fake_redis(TestContextDecorator):
    """Replace our prometheus redis instance with a fakeredis."""

    @override
    def __init__(self, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        self.settings_override = override_settings(
            PROMETHEUS_REDIS_URL="fakeredis:///2"
        )

    @override
    def enable(self) -> None:
        self.settings_override.enable()
        self._old_daemon_threads = _daemon_threads.copy()
        _daemon_threads.clear()
        self._live_metrics = _live_metrics.copy()
        _live_metrics.clear()
        fake_client = redis_client()
        fake_client.flushdb()

    @override
    def disable(self) -> None:
        for identifier in list(_daemon_threads):
            mark_process_dead(identifier)
        _daemon_threads.clear()
        _daemon_threads.update(self._old_daemon_threads)
        _live_metrics.clear()
        _live_metrics.update(self._live_metrics)
        self.settings_override.disable()
