from __future__ import annotations

from collections.abc import Callable
from functools import partial, wraps
from typing import Any

from flask import request
from werkzeug.exceptions import NotFound, Unauthorized

from configs import dify_config
from controllers.openapi.auth.context import Context
from controllers.openapi.auth.pipelines import pipeline_for_subject
from controllers.openapi.auth.requirements import assert_license_valid
from controllers.openapi.auth.resource_access import authenticate_resource_token
from controllers.openapi.auth.spec import EndpointSpec
from controllers.openapi.auth.subjects import subject_from_auth
from core.db.session_factory import session_factory
from enums import DeploymentEdition
from libs.oauth_bearer import (
    InvalidBearerError,
    TokenType,
    assert_bearer_feature_enabled,
    extract_bearer,
    get_authenticator,
)


class AuthRouter:
    def guard(self, spec: EndpointSpec) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
        def decorator(view: Callable[..., Any]) -> Callable[..., Any]:
            @wraps(view)
            def decorated(*args: Any, **kwargs: Any) -> Any:
                return self._execute(spec, partial(view, *args, **kwargs))

            return decorated

        return decorator

    def _execute(self, spec: EndpointSpec, call: Callable[..., Any]) -> Any:
        """The order is the contract. An endpoint the edition does not expose
        answers 404 before anything reveals whether the bearer was valid, and
        an enterprise deployment's licence answers 403 before the missing-bearer
        401. The licence is a fact about the deployment, not the route or the
        caller, so this is the one place it is checked. The OAuth feature flag
        answers 503 before OAuth authentication, but does not gate independent
        resource credentials. Missing or unknown credentials still receive 503
        when OAuth is disabled.
        """
        if spec.edition is not None and dify_config.DEPLOYMENT_EDITION not in spec.edition:
            raise NotFound()
        if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.ENTERPRISE:
            assert_license_valid()
        token = extract_bearer(request)
        is_resource_token = token is not None and TokenType.for_token(token) is TokenType.RESOURCE_ACCESS
        if not is_resource_token:
            assert_bearer_feature_enabled()
        if not token:
            raise Unauthorized("bearer required")

        try:
            auth = authenticate_resource_token(token) if is_resource_token else get_authenticator().authenticate(token)
        except InvalidBearerError:
            # One answer for every rejection reason - unknown prefix, no live row,
            # expired - so a caller cannot probe which one it hit. Same reasoning as
            # the 404-not-403 elsewhere on this surface.
            raise Unauthorized("invalid bearer")
        subject = subject_from_auth(auth)
        pipeline = pipeline_for_subject(subject)

        # One session spans the loaders and the handler, so the objects the
        # requirements load stay attached while the handler reads them. It is
        # finalised here, like `with_session` does, until repositories own it.
        with session_factory.create_session() as session:
            ctx = Context(subject, session, dict(request.view_args or {}))
            try:
                result = pipeline.run(
                    subject=subject,
                    auth=auth,
                    spec=spec,
                    ctx=ctx,
                    session=session,
                    call=call,
                )
            except Exception:
                session.rollback()  # guard-ignore: no-new-controller-sqlalchemy -- the router owns the rollback
                raise
            session.commit()
            return result


subject_router = AuthRouter()