feat(events): extract event information

hathor/event/__init__.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from hathor.event.base_event import BaseEvent
 from hathor.event.event_manager import EventManager
 
-__all__ = ['BaseEvent', 'EventManager']
+__all__ = ['EventManager']

hathor/event/event_manager.py

@@ -12,11 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Any, Callable, Dict, Optional
+from typing import Callable, Optional
 
 from structlog import get_logger
 
-from hathor.event.base_event import BaseEvent
+from hathor.event.model.base_event import BaseEvent
 from hathor.event.storage import EventStorage
 from hathor.event.websocket import EventWebsocketFactory
 from hathor.pubsub import EventArguments, HathorEvents, PubSubManager
@@ -50,48 +50,6 @@
 }
 
 
-def _empty(args: EventArguments) -> Dict[str, Any]:
-    return {}
-
-
-def _extract_tx(args: EventArguments) -> Dict[str, Any]:
-    return {
-        'hash': args.tx.hash_hex,
-        # TODO: other fields haven't been implemented, but will be before this feature is rolled out
-    }
-
-
-def _extract_reorg(args: EventArguments) -> Dict[str, Any]:
-    return {
-        'reorg_size': args.reorg_size,
-        'previous_best_block': args.old_best_block.hash_hex,
-        'new_best_block': args.new_best_block.hash_hex,
-        'common_block': args.common_block.hash_hex,
-    }
-
-
-_EVENT_EXTRACT_MAP: Dict[HathorEvents, Callable[[EventArguments], Dict[str, Any]]] = {
-    HathorEvents.LOAD_STARTED: _empty,
-    HathorEvents.LOAD_FINISHED: _empty,
-    HathorEvents.NETWORK_NEW_TX_ACCEPTED: _extract_tx,
-    HathorEvents.NETWORK_BEST_BLOCK_FOUND: _extract_tx,
-    HathorEvents.NETWORK_ORPHAN_BLOCK_FOUND: _extract_tx,
-    HathorEvents.REORG_STARTED: _extract_reorg,
-    HathorEvents.REORG_FINISHED: _empty,
-    HathorEvents.VERTEX_METADATA_CHANGED: _extract_tx,
-    HathorEvents.CONSENSUS_TX_UPDATE: _extract_tx,
-    HathorEvents.CONSENSUS_TX_REMOVED: _extract_tx,
-}
-
-
-def _build_event_data(event_type: HathorEvents, event_args: EventArguments) -> Dict[str, Any]:
-    """Extract and build event data from event_args for a given event type."""
-    event_extract_fn = _EVENT_EXTRACT_MAP.get(event_type)
-    if event_extract_fn is None:
-        raise ValueError(f'The given event type ({event_type}) is not a supported event')
-    return event_extract_fn(event_args)
-
-
 class EventManager:
     """Class that manages integration events.
 
@@ -211,11 +169,11 @@ def _create_event(
         event_args: EventArguments,
         group_id: Optional[int],
     ) -> BaseEvent:
-        return BaseEvent(
-            id=0 if self._last_event is None else self._last_event.id + 1,
+        return BaseEvent.from_event_arguments(
+            event_id=0 if self._last_event is None else self._last_event.id + 1,
             peer_id=self._peer_id,
             timestamp=self._clock.seconds(),
-            type=event_type.value,
-            data=_build_event_data(event_type, event_args),
+            event_type=event_type,
+            event_args=event_args,
             group_id=group_id,
         )

hathor/event/model/base_event.py

@@ -0,0 +1,86 @@
+# Copyright 2022 Hathor Labs
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import Dict, Optional, Type
+
+from pydantic import NonNegativeInt, validator
+
+from hathor.event.model.event_data import BaseEventData, EmptyData, EventData, ReorgData, TxData
+from hathor.pubsub import EventArguments, HathorEvents
+from hathor.utils.pydantic import BaseModel
+
+_EVENT_DATA_MAP: Dict[HathorEvents, Type[BaseEventData]] = {
+    HathorEvents.LOAD_STARTED: EmptyData,
+    HathorEvents.LOAD_FINISHED: EmptyData,
+    HathorEvents.NETWORK_NEW_TX_ACCEPTED: TxData,
+    HathorEvents.NETWORK_BEST_BLOCK_FOUND: TxData,
+    HathorEvents.NETWORK_ORPHAN_BLOCK_FOUND: TxData,
+    HathorEvents.REORG_STARTED: ReorgData,
+    HathorEvents.REORG_FINISHED: EmptyData,
+    HathorEvents.VERTEX_METADATA_CHANGED: TxData,
+    HathorEvents.CONSENSUS_TX_UPDATE: TxData,
+    HathorEvents.CONSENSUS_TX_REMOVED: TxData,
+}
+
+
+class BaseEvent(BaseModel, use_enum_values=True):
+    # Full node id, because different full nodes can have different sequences of events
+    peer_id: str
+    # Event unique id, determines event order
+    id: NonNegativeInt
+    # Timestamp in which the event was emitted, this follows the unix_timestamp format, it's only informative, events
+    # aren't guaranteed to always have sequential timestamps, for example, if the system clock changes between two
+    # events it's possible that timestamps will temporarily decrease.
+    timestamp: float
+    # One of the event types
+    type: HathorEvents
+    # Variable for event type
+    data: EventData
+    # Used to link events, for example, many TX_METADATA_CHANGED will have the same group_id when they belong to the
+    # same reorg process
+    group_id: Optional[NonNegativeInt] = None
+
+    @classmethod
+    def from_event_arguments(
+        cls,
+        peer_id: str,
+        event_id: NonNegativeInt,
+        timestamp: float,
+        event_type: HathorEvents,
+        event_args: EventArguments,
+        group_id: Optional[NonNegativeInt]
+    ) -> 'BaseEvent':
+        event_data_type = _EVENT_DATA_MAP.get(event_type)
+
+        if event_data_type is None:
+            raise ValueError(f'The given event type ({event_type}) is not a supported event')
+
+        return cls(
+            peer_id=peer_id,
+            id=event_id,
+            timestamp=timestamp,
+            type=event_type,
+            data=event_data_type.from_event_arguments(event_args),
+            group_id=group_id,
+        )
+
+    @validator('data')
+    def data_type_must_match_event_type(cls, v, values):
+        event_type = HathorEvents(values['type'])
+        expected_data_type = _EVENT_DATA_MAP.get(event_type)
+
+        if type(v) != expected_data_type:
+            raise ValueError('event data type does not match event type')
+
+        return v

hathor/event/model/event_data.py

@@ -0,0 +1,109 @@
+#  Copyright 2023 Hathor Labs
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from typing import List, Optional, Union
+
+from pydantic import Extra
+
+from hathor.pubsub import EventArguments
+from hathor.utils.pydantic import BaseModel
+
+
+class TxInput(BaseModel):
+    tx_id: str
+    index: int
+    data: int
+
+
+class TxOutput(BaseModel):
+    value: int
+    script: str
+    token_data: int
+
+
+class SpentOutput(BaseModel):
+    index: int
+    tx_ids: List[str]
+
+
+class SpentOutputs(BaseModel):
+    spent_output: List[SpentOutput]
+
+
+class TxMetadata(BaseModel, extra=Extra.ignore):
+    hash: str
+    spent_outputs: List[SpentOutputs]
+    conflict_with: List[str]
+    voided_by: List[str]
+    received_by: List[int]
+    children: List[str]
+    twins: List[str]
+    accumulated_weight: float
+    score: float
+    first_block: Optional[str]
+    height: int
+    validation: str
+
+
+class BaseEventData(BaseModel):
+    @classmethod
+    def from_event_arguments(cls, args: EventArguments) -> 'EventData':
+        raise NotImplementedError()
+
+
+class EmptyData(BaseEventData):
+    @classmethod
+    def from_event_arguments(cls, args: EventArguments) -> 'EmptyData':
+        return cls()
+
+
+class TxData(BaseEventData, extra=Extra.ignore):
+    hash: str
+    nonce: int
+    timestamp: int
+    version: int
+    weight: float
+    inputs: List['TxInput']
+    outputs: List['TxOutput']
+    parents: List[str]
+    tokens: List[str]
+    # TODO: Token name and symbol could be in a different class because they're only used by TokenCreationTransaction
+    token_name: Optional[str]
+    token_symbol: Optional[str]
+    metadata: 'TxMetadata'
+
+    @classmethod
+    def from_event_arguments(cls, args: EventArguments) -> 'TxData':
+        tx_json = args.tx.to_json(include_metadata=True)
+
+        return cls(**tx_json)
+
+
+class ReorgData(BaseEventData):
+    reorg_size: int
+    previous_best_block: str
+    new_best_block: str
+    common_block: str
+
+    @classmethod
+    def from_event_arguments(cls, args: EventArguments) -> 'ReorgData':
+        return cls(
+            reorg_size=args.reorg_size,
+            previous_best_block=args.old_best_block.hash_hex,
+            new_best_block=args.new_best_block.hash_hex,
+            common_block=args.common_block.hash_hex,
+        )
+
+
+EventData = Union[EmptyData, TxData, ReorgData]

hathor/event/resources/event.py

@@ -20,7 +20,8 @@
 from hathor.api_util import Resource, set_cors
 from hathor.cli.openapi_files.register import register_resource
 from hathor.conf import HathorSettings
-from hathor.event import BaseEvent, EventManager
+from hathor.event import EventManager
+from hathor.event.model.base_event import BaseEvent
 from hathor.utils.api import ErrorResponse, QueryParams, Response
 
 settings = HathorSettings()

hathor/event/storage/event_storage.py

@@ -15,7 +15,7 @@
 from abc import ABC, abstractmethod
 from typing import Iterator, Optional
 
-from hathor.event.base_event import BaseEvent
+from hathor.event.model.base_event import BaseEvent
 
 
 class EventStorage(ABC):

hathor/event/storage/memory_storage.py

@@ -14,7 +14,7 @@
 
 from typing import Iterator, List, Optional
 
-from hathor.event.base_event import BaseEvent
+from hathor.event.model.base_event import BaseEvent
 from hathor.event.storage.event_storage import EventStorage
 
 

hathor/event/storage/rocksdb_storage.py

@@ -14,7 +14,7 @@
 
 from typing import Iterator, Optional
 
-from hathor.event.base_event import BaseEvent
+from hathor.event.model.base_event import BaseEvent
 from hathor.event.storage.event_storage import EventStorage
 from hathor.storage.rocksdb_storage import RocksDBStorage
 from hathor.transaction.util import int_to_bytes

hathor/event/websocket/factory.py

@@ -17,7 +17,7 @@
 from autobahn.twisted.websocket import WebSocketServerFactory
 from structlog import get_logger
 
-from hathor.event import BaseEvent
+from hathor.event.model.base_event import BaseEvent
 from hathor.event.storage import EventStorage
 from hathor.event.websocket.protocol import EventWebsocketProtocol
 from hathor.event.websocket.response import EventResponse, InvalidRequestType

hathor/event/websocket/response.py

@@ -16,7 +16,7 @@
 
 from pydantic import Field, NonNegativeInt
 
-from hathor.event import BaseEvent
+from hathor.event.model.base_event import BaseEvent
 from hathor.utils.pydantic import BaseModel