# Source code for nio.events.ephemeral

    # -*- coding: utf-8 -*-

    # Copyright © 2019 Damir Jelić <poljar@termina.org.uk>
    #
    # Permission to use, copy, modify, and/or distribute this software for
    # any purpose with or without fee is hereby granted, provided that the
    # above copyright notice and this permission notice appear in all copies.
    #
    # THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
    # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
    # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
    # SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
    # RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
    # CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
    # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

    """nio Ephemeral events.

    Ephemeral events are a special type of events that are not recorded in the room
    history.

    Ephemeral events are used for typing notifications and read receipts.

    """

    from typing import List, Dict

    from dataclasses import dataclass, field

    from ..schemas import Schemas
    from .misc import verify_or_none

    @dataclass
    class EphemeralEvent:
        """Base class for ephemeral events."""

        @classmethod
        @verify_or_none(Schemas.ephemeral_event)
        def parse_event(cls, event_dict):
            """Parse an ephemeral event and create a higher level event object.

            This function parses the type of the ephemeral event and produces a
            higher level event object representing the parsed event.

            The event structure is checked for correctness and the event fields are
            type-checked. If this validation process fails for an event None will
            be returned.

            If the event has an unknown type None is returned as well.

            Args:
                event_dict (dict): The dictionary representation of the event.

            """
            if event_dict["type"] == "m.typing":
                return TypingNoticeEvent.from_dict(event_dict)
            if event_dict["type"] == "m.receipt":
                return ReceiptEvent.from_dict(event_dict)

            return None

        @classmethod
        def from_dict(cls, parsed_dict):
            """Create an Ephemeral event from a dictionary.

            Args:
                parsed_dict (dict): The dictionary representation of the event.

            """
            raise NotImplementedError()

    @dataclass
    class TypingNoticeEvent(EphemeralEvent):
        """Informs the client of the list of users currently typing in a room.

        Attributes:
            users (List): The list of user IDs typing in this room, if any.

        """

        users: List = field()

        @classmethod
        @verify_or_none(Schemas.m_typing)
        def from_dict(cls, parsed_dict):
            return cls(parsed_dict["content"]["user_ids"])

    @dataclass
    class Receipt:
        """Receipt of a user acknowledging an event.

        If `receipt_type` is "m.read", then it is a read receipt and shows the last
        event that a user has read.

        Attributes:
            event_id (str): the ID of the event being acknowleged
            receipt_type (str): the type of receipt being received; this is
                commonly "m.read" for read receipts.
            user_id (str): the ID of the user who is acknowledging the event.
            timestamp (int): The timestamp the receipt was sent at.
        """
        event_id: str = field()
        receipt_type: str = field()
        user_id: str = field()
        timestamp: int = field()

    @dataclass
    class ReceiptEvent(EphemeralEvent):
        """Informs the client of changes in the newest events seen by users.

        A ReceiptEvent can contain multiple event_ids seen by many different users.
        At the time of writing, all Receipts have a `receipt_type` of "m.read" and
        are read receipts, but this may change in the future.

        Attributes:
            receipts (List[Receipt]): The list of `Receipt`s in this event.
        """
        receipts: List[Receipt] = field()

        @classmethod
        @verify_or_none(Schemas.m_receipt)
        def from_dict(cls, parsed_dict):
            event_receipts: List[Receipt] = []

            for event_id, event in parsed_dict["content"].items():
                for receipt_type, receipt in event.items():
                    for user_id, user in receipt.items():
                        # Synapse pre-0.99.3 has a bug where it sends invalid
                        # ts values. https://github.com/matrix-org/synapse/issues/4898
                        if isinstance(user, dict) and "ts" in user:
                            event_receipts.append(
                                Receipt(event_id, receipt_type, user_id, user["ts"])
                            )

            return cls(event_receipts)