privaterelay/utils.py

from __future__ import annotations

import json
import logging
import random
from collections.abc import Callable
from decimal import Decimal
from functools import cache, wraps
from pathlib import Path
from typing import TYPE_CHECKING, ParamSpec, TypedDict, TypeVar, cast

from django.conf import settings
from django.contrib.auth.models import AbstractBaseUser
from django.http import Http404, HttpRequest

from waffle import get_waffle_flag_model
from waffle.models import logger as waffle_logger
from waffle.utils import get_cache as get_waffle_cache
from waffle.utils import get_setting as get_waffle_setting

from privaterelay.country_utils import (
    AcceptLanguageError,
    _get_cc_from_lang,
    _get_cc_from_request,
    guess_country_from_accept_lang,
)
from privaterelay.sp3_plans import SP3PlanCountryLangMapping

from .plans import (
    CountryStr,
    LanguageStr,
    PeriodStr,
    PlanCountryLangMapping,
    get_premium_country_language_mapping,
)

if TYPE_CHECKING:
    from .glean_interface import RelayGleanLogger

info_logger = logging.getLogger("eventsinfo")


class CountryInfo(TypedDict):
    country_code: str
    countries: list[CountryStr]
    available_in_country: bool
    plan_country_lang_mapping: PlanCountryLangMapping | SP3PlanCountryLangMapping


def get_countries_info_from_request_and_mapping(
    request: HttpRequest, mapping: PlanCountryLangMapping | SP3PlanCountryLangMapping
) -> CountryInfo:
    country_code = _get_cc_from_request(request)
    countries = sorted(mapping.keys())
    available_in_country = country_code in countries
    return {
        "country_code": country_code,
        "countries": countries,
        "available_in_country": available_in_country,
        "plan_country_lang_mapping": mapping,
    }


def get_countries_info_from_lang_and_mapping(
    accept_lang: str, mapping: PlanCountryLangMapping
) -> CountryInfo:
    country_code = _get_cc_from_lang(accept_lang)
    countries = sorted(mapping.keys())
    available_in_country = country_code in countries
    return {
        "country_code": country_code,
        "countries": countries,
        "available_in_country": available_in_country,
        "plan_country_lang_mapping": mapping,
    }


def get_subplat_upgrade_link_by_language(
    accept_language: str, period: PeriodStr = "yearly"
) -> str:
    try:
        country_str = guess_country_from_accept_lang(accept_language)
        country = cast(CountryStr, country_str)
    except AcceptLanguageError:
        country = "US"
    language_str = accept_language.split("-")[0].lower()
    language = cast(LanguageStr, language_str)
    country_lang_mapping = get_premium_country_language_mapping()
    country_details = country_lang_mapping.get(country, country_lang_mapping["US"])
    if language in country_details:
        plan = country_details[language][period]
    else:
        first_key = list(country_details.keys())[0]
        plan = country_details[first_key][period]
    return (
        f"{settings.FXA_BASE_ORIGIN}/subscriptions/products/"
        f"{settings.PERIODICAL_PREMIUM_PROD_ID}?plan={plan['id']}"
    )


# Generics for defining function decorators
# https://mypy.readthedocs.io/en/stable/generics.html#declaring-decorators
_Params = ParamSpec("_Params")
_RetVal = TypeVar("_RetVal")


def enable_or_404(
    check_function: Callable[[], bool],
    message: str = "This conditional view is disabled.",
) -> Callable[[Callable[_Params, _RetVal]], Callable[_Params, _RetVal]]:
    """
    Returns decorator that enables a view if a check function passes,
    otherwise returns a 404.

    Usage:

        def percent_1():
           import random
           return random.randint(1, 100) == 1

        @enable_if(percent_1)
        def lucky_view(request):
            #  1 in 100 chance of getting here
            # 99 in 100 chance of 404
    """

    def decorator(func: Callable[_Params, _RetVal]) -> Callable[_Params, _RetVal]:
        @wraps(func)
        def inner(*args: _Params.args, **kwargs: _Params.kwargs) -> _RetVal:
            if check_function():
                return func(*args, **kwargs)
            else:
                raise Http404(message)  # Display a message with DEBUG=True

        return inner

    return decorator


def enable_if_setting(
    setting_name: str,
    message_fmt: str = "This view is disabled because {setting_name} is False",
) -> Callable[[Callable[_Params, _RetVal]], Callable[_Params, _RetVal]]:
    """
    Returns decorator that enables a view if a setting is truthy, otherwise
    returns a 404.

    Usage:

        @enable_if_setting("DEBUG")
        def debug_only_view(request):
            # DEBUG == True

    Or in URLS:

        path(
            "developer_info",
            enable_if_setting("DEBUG")(debug_only_view)
        ),
        name="developer-info",
    ),

    """

    def setting_is_truthy() -> bool:
        return bool(getattr(settings, setting_name))

    return enable_or_404(
        setting_is_truthy, message_fmt.format(setting_name=setting_name)
    )


def flag_is_active_in_task(flag_name: str, user: AbstractBaseUser | None) -> bool:
    """
    Test if a flag is active in a task (not in a web request).

    This mirrors AbstractBaseFlag.is_active, replicating these checks:
    * Logs missing flags, if configured
    * Creates missing flags, if configured
    * Returns default for missing flags
    * Checks flag.everyone
    * Checks flag.users and flag.groups, if a user is passed
    * Returns random results for flag.percent

    It does not include:
    * Overriding a flag with a query parameter
    * Persisting a flag in a cookie (includes percent flags)
    * Language-specific overrides (could be added later)
    * Read-only mode for percent flags

    When using this function, use the @override_flag decorator in tests, rather
    than manually creating flags in the database.
    """
    flag = get_waffle_flag_model().get(flag_name)
    if not flag.pk:
        log_level = get_waffle_setting("LOG_MISSING_FLAGS")
        if log_level:
            waffle_logger.log(log_level, "Flag %s not found", flag_name)
        if get_waffle_setting("CREATE_MISSING_FLAGS"):
            flag, _created = get_waffle_flag_model().objects.get_or_create(
                name=flag_name,
                defaults={"everyone": get_waffle_setting("FLAG_DEFAULT")},
            )
            cache = get_waffle_cache()
            cache.set(flag._cache_key(flag.name), flag)

        return bool(get_waffle_setting("FLAG_DEFAULT"))

    # Removed - check for override as request query parameter

    if flag.everyone:
        return True
    elif flag.everyone is False:
        return False

    # Removed - check for testing override in request query or cookie
    # Removed - check for language-specific override

    if user is not None:
        active_for_user = flag.is_active_for_user(user)
        if active_for_user is not None:
            return bool(active_for_user)

    if flag.percent and flag.percent > 0:
        # Removed - check for waffles attribute of request
        # Removed - check for cookie setting for flag
        # Removed - check for read-only mode

        if Decimal(str(random.uniform(0, 100))) <= flag.percent:  # noqa: S311
            # Removed - setting the flag for future checks
            return True

    return False


class VersionInfo(TypedDict):
    source: str
    version: str
    commit: str
    build: str


@cache
def get_version_info(base_dir: str | Path | None = None) -> VersionInfo:
    """Return version information written by build process."""
    if base_dir is None:
        base_path = Path(settings.BASE_DIR)
    else:
        base_path = Path(base_dir)
    version_json_path = base_path / "version.json"
    info = {}
    if version_json_path.exists():
        with version_json_path.open() as version_file:
            try:
                info = json.load(version_file)
            except ValueError:
                pass
            if not hasattr(info, "get"):
                info = {}
    version_info = VersionInfo(
        source=info.get("source", "https://github.com/mozilla/fx-private-relay"),
        version=info.get("version", "unknown"),
        commit=info.get("commit", "unknown"),
        build=info.get("build", "not built"),
    )
    return version_info


@cache
def glean_logger() -> RelayGleanLogger:
    from .glean_interface import RelayGleanLogger

    version_info = get_version_info()
    return RelayGleanLogger(
        application_id="relay-backend",
        app_display_version=version_info["version"],
        channel=settings.RELAY_CHANNEL,
    )