# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0

from collections.abc import Mapping, Sequence
from logging import getLogger
from threading import Lock
from time import time_ns
from types import NoneType
from typing import TYPE_CHECKING, cast

from typing_extensions import assert_never

from opentelemetry.sdk.metrics._internal.aggregation import (
    Aggregation,
    AggregationTemporality,
    DefaultAggregation,
    _Aggregation,
    _SumAggregation,
)
from opentelemetry.sdk.metrics._internal.instrument import _Instrument
from opentelemetry.sdk.metrics._internal.measurement import Measurement
from opentelemetry.sdk.metrics._internal.point import DataPointT
from opentelemetry.sdk.metrics._internal.view import View
from opentelemetry.util.types import AnyValue, Attributes

_logger = getLogger(__name__)

# Every branch is tagged so that values Python considers equal but the OTel
# data model does not -- True/1/1.0, or a sequence of pairs and the mapping
# it resembles -- produce different aggregation keys.
_HashedAttributes = tuple[
    str,
    str
    | bool
    | int
    | float
    | bytes
    | None
    | tuple["_HashedAttributes", ...]
    | tuple[tuple[str, "_HashedAttributes"], ...],
]


# pylint: disable=inconsistent-return-statements
def _hash_attributes(value: Attributes | AnyValue) -> _HashedAttributes:
    # Attributes have been cleaned and validated when Measurement was instantiated,
    # so value is guaranteed to match one of the branches below at runtime.
    if isinstance(value, (NoneType, str, int, float, bool, bytes)):
        # bool is a subclass of int and 1 == 1.0, so the value alone is not
        # enough to tell these apart.
        return (type(value).__name__, value)
    if isinstance(value, Sequence):
        return ("sequence", tuple(_hash_attributes(v) for v in value))
    if isinstance(value, Mapping):
        return (
            "mapping",
            tuple(
                (k, _hash_attributes(value[k]))
                for k in sorted(
                    value,
                    key=lambda item: item if isinstance(item, str) else str(item),
                )
            ),
        )
    if TYPE_CHECKING:
        assert_never(value)


_DEFAULT_CARDINALITY_LIMIT = 2000

_OVERFLOW_ATTRIBUTES: dict[str, AnyValue] = {"otel.metric.overflow": True}
_OVERFLOW_KEY = _hash_attributes(_OVERFLOW_ATTRIBUTES)


class _ViewInstrumentMatch:
    def __init__(
        self,
        view: View,
        instrument: _Instrument,
        instrument_class_aggregation: dict[type, Aggregation],
        cardinality_limit: int = _DEFAULT_CARDINALITY_LIMIT,
        reset_on_collect: bool = False,
    ):
        self._view = view
        self._cardinality_limit = cardinality_limit
        self._reset_on_collect = reset_on_collect
        # Delta aggregations created after a collection continue from it.
        self._previous_collection_start_nanos: int | None = None
        self._instrument = instrument
        self._attributes_aggregation: dict[_HashedAttributes, _Aggregation] = {}
        self._lock = Lock()
        self._instrument_class_aggregation = instrument_class_aggregation
        self._name = self._view._name or self._instrument.name
        self._description = self._view._description or self._instrument.description
        if not isinstance(self._view._aggregation, DefaultAggregation):
            self._aggregation = self._view._aggregation._create_aggregation(
                self._instrument,
                None,
                self._view._exemplar_reservoir_factory,
                0,
            )
        else:
            self._aggregation = self._instrument_class_aggregation[self._instrument.__class__]._create_aggregation(
                self._instrument,
                None,
                self._view._exemplar_reservoir_factory,
                0,
            )

    def conflicts(self, other: "_ViewInstrumentMatch") -> bool:
        # pylint: disable=protected-access

        result = (
            self._name == other._name
            and self._instrument.unit == other._instrument.unit
            # The aggregation class is being used here instead of data point
            # type since they are functionally equivalent.
            and self._aggregation.__class__ == other._aggregation.__class__
        )
        if not result:
            return result

        if isinstance(self._aggregation, _SumAggregation):
            # if result is True the two aggregation are of the same type
            self._aggregation = cast(_SumAggregation, self._aggregation)
            other._aggregation = cast(_SumAggregation, other._aggregation)

            result = (
                self._aggregation._instrument_is_monotonic == other._aggregation._instrument_is_monotonic
                and self._aggregation._instrument_aggregation_temporality
                == other._aggregation._instrument_aggregation_temporality
            )

        return result

    # pylint: disable=protected-access
    def consume_measurement(self, measurement: Measurement, should_sample_exemplar: bool = True) -> None:
        attributes = {}
        if measurement.attributes:
            # Make a shallow copy since the user can mutate the dict after the fact.
            # The user can still modify mutable attribute values (lists/dicts)
            # leading to unexpected behavior, but deep copying is expensive.
            attributes = dict(measurement.attributes)
        if self._view._attribute_keys is not None:
            attributes = {k: v for k, v in attributes.items() if k in self._view._attribute_keys}

        aggr_key = _hash_attributes(attributes)

        if self._reset_on_collect:
            # collect() discards the aggregations, so hold the lock until the
            # measurement is aggregated to not lose it.
            with self._lock:
                self._get_or_create_aggregation(aggr_key, attributes).aggregate(measurement, should_sample_exemplar)
            return

        aggregation = self._attributes_aggregation.get(aggr_key)
        if aggregation is None:
            with self._lock:
                aggregation = self._get_or_create_aggregation(aggr_key, attributes)

        aggregation.aggregate(measurement, should_sample_exemplar)

    def _get_or_create_aggregation(self, aggr_key: _HashedAttributes, attributes: Attributes) -> _Aggregation:
        """Must be called with ``self._lock`` held."""
        aggregation = self._attributes_aggregation.get(aggr_key)
        if aggregation is not None:
            return aggregation

        # Overflow is only used once the limit of distinct attribute sets is
        # reached, so the overflow point is exported in addition to them.
        if len(self._attributes_aggregation) >= self._cardinality_limit:
            aggr_key = _OVERFLOW_KEY
            attributes = dict(_OVERFLOW_ATTRIBUTES)
            aggregation = self._attributes_aggregation.get(aggr_key)
            if aggregation is not None:
                return aggregation

        start_time_unix_nano = self._previous_collection_start_nanos or time_ns()
        # pylint: disable=protected-access
        if not isinstance(self._view._aggregation, DefaultAggregation):
            aggregation = self._view._aggregation._create_aggregation(
                self._instrument,
                attributes,
                self._view._exemplar_reservoir_factory,
                start_time_unix_nano,
            )
        else:
            aggregation = self._instrument_class_aggregation[self._instrument.__class__]._create_aggregation(
                self._instrument,
                attributes,
                self._view._exemplar_reservoir_factory,
                start_time_unix_nano,
            )
        self._attributes_aggregation[aggr_key] = aggregation
        return aggregation

    def collect(
        self,
        collection_aggregation_temporality: AggregationTemporality,
        collection_start_nanos: int,
    ) -> Sequence[DataPointT] | None:
        data_points: list[DataPointT] = []
        with self._lock:
            aggregations = self._attributes_aggregation
            if self._reset_on_collect and collection_aggregation_temporality is AggregationTemporality.DELTA:
                self._attributes_aggregation = {}
                self._previous_collection_start_nanos = collection_start_nanos

            for aggregation in aggregations.values():
                data_point = aggregation.collect(collection_aggregation_temporality, collection_start_nanos)
                if data_point is not None:
                    data_points.append(data_point)

        # Returning here None instead of an empty list because the caller
        # does not consume a sequence and to be consistent with the rest of
        # collect methods that also return None.
        return data_points or None