feat: add common main

Author: emptybutton

deploy/dev/tgdb/conf.yaml

@@ -7,6 +7,14 @@ conf:
     bots: "/app/deploy/dev/tgdb/bots/bots.txt"
     userbots: "/app/deploy/dev/tgdb/bots/userbots.txt"
 
+  horizon:
+    max_len: 10_000
+    transaction:
+      max_age_seconds: 10
+
+  message_cache:
+    max_len: 1_000_000
+
   heap:
     chat: -1005010263669
     page:
@@ -15,6 +23,7 @@ conf:
   relations:
     chat: -1005000098156
     vacuum:
+      max_workers: 1
       window_delay_seconds: 30
       min_message_count: 1
 
@@ -25,5 +34,6 @@ conf:
       timeout_seconds: 0.1
 
     vacuum:
+      max_workers: 10
       window_delay_seconds: 10
       min_message_count: 200

src/tgdb/infrastructure/adapters/buffer.py

@@ -14,8 +14,8 @@
 
 @dataclass(frozen=True, unsafe_hash=False)
 class InMemoryBuffer[ValueT](Buffer[ValueT]):
-    _size_to_overflow: int
-    _overflow_timeout_seconds: int
+    _len_to_overflow: int
+    _overflow_timeout_seconds: int | float
     _values: deque[ValueT]
     _is_overflowed: Event = field(init=False, default_factory=Event)
 
@@ -36,11 +36,11 @@ async def __aiter__(self) -> AsyncIterator[Sequence[ValueT]]:
 
             yield tuple(
                 self._values.popleft()
-                for _ in range(self._size_to_overflow)
+                for _ in range(self._len_to_overflow)
             )
 
     def _refresh_overflow(self) -> None:
-        if len(self._values) >= self._size_to_overflow:
+        if len(self._values) >= self._len_to_overflow:
             self._is_overflowed.set()
 
 

src/tgdb/infrastructure/adapters/relations.py

@@ -41,7 +41,7 @@ async def add(self, relation: Relation) -> None:
 
 @dataclass
 class InTelegramReplicableRelations(Relations):
-    _in_telegram_encoded_relations: InTelegramBytes
+    _in_tg_encoded_relations: InTelegramBytes
     _cached_relations: InMemoryDb[Relation]
 
     async def __aenter__(self) -> Self:
@@ -86,10 +86,10 @@ async def add(self, relation: Relation) -> None:
         self._cached_relations.insert(relation)
 
         encoded_cached_relations = pickle.dumps(tuple(self._cached_relations))
-        await self._in_telegram_encoded_relations.set(encoded_cached_relations)
+        await self._in_tg_encoded_relations.set(encoded_cached_relations)
 
     async def _loaded_relations(self) -> tuple[Relation, ...]:
-        encoded_relations = await self._in_telegram_encoded_relations
+        encoded_relations = await self._in_tg_encoded_relations
 
         if encoded_relations is None:
             return tuple()

src/tgdb/infrastructure/adapters/shared_horizon.py

@@ -1,9 +1,11 @@
+from dataclasses import dataclass
 from types import TracebackType
 
 from tgdb.application.common.ports.shared_horizon import SharedHorizon
 from tgdb.entities.horizon.horizon import Horizon
 
 
+@dataclass(frozen=True)
 class InMemorySharedHorizon(SharedHorizon):
     _horizon: Horizon
 

src/tgdb/infrastructure/lazy_map.py

@@ -37,18 +37,23 @@ async def __getitem__(self, key: KeyT) -> ValueT:
             return self._output(self._cache_map[key])
 
         value = await self._external_value(key)
-        self._cache_map[key] = value
-
-        if len(self._cache_map) > self._cache_map_max_len:
-            self._cache_map.pop(next(iter(self._cache_map)))
+        self._insert_to_cache_map(key, value)
 
         return self._output(value)
 
     def __setitem__(self, key: KeyT, value: ValueT) -> None:
-        self._cache_map[key] = value
+        self._insert_to_cache_map(key, value)
 
     def _output(self, value: ExternalValue[ValueT]) -> ValueT:
         if isinstance(value, NoExternalValue):
             raise KeyError
 
         return value
+
+    def _insert_to_cache_map(
+        self, key: KeyT, value: ExternalValue[ValueT]
+    ) -> None:
+        self._cache_map[key] = value
+
+        if len(self._cache_map) > self._cache_map_max_len:
+            self._cache_map.pop(next(iter(self._cache_map)))

src/tgdb/infrastructure/pyyaml/conf.py

@@ -2,53 +2,76 @@
 from pathlib import Path
 
 import yaml
+from pydantic import BaseModel, Field
 
 
-@dataclass(frozen=True)
-class Conf:
-    api_id: int
-    api_hash: str
+class APIConf(BaseModel):
+    id: int
+    hash: str
 
-    clients_bots: str
-    clients_userbots: str
 
-    heap_chat: int
-    heap_page_fullness: float
+class ClientsConf(BaseModel):
+    bots: Path
+    userbots: Path
 
-    relations_chat: int
-    relations_vacuum_window_delay_seconds: int
-    relations_vacuum_min_message_count: int
 
-    buffer_chat: int
-    buffer_overflow_len: int
-    buffer_overflow_timeout_seconds: float
-    buffer_vacuum_window_delay_seconds: int
-    buffer_vacuum_min_message_count: int
+class TransactionConf(BaseModel):
+    max_age_seconds: int = Field(..., alias="max_age_seconds")
 
-    @classmethod
-    def load(cls, path: Path) -> "Conf":
-        with path.open("r") as file:
-            data = yaml.safe_load(file)
 
-        conf = data["conf"]
+class HorizonConf(BaseModel):
+    max_len: int
+    transaction: TransactionConf
+
+
+class MessageCacheConf(BaseModel):
+    max_len: int
+
+
+class PageConf(BaseModel):
+    fullness: float
+
+
+class HeapConf(BaseModel):
+    chat: int
+    page: PageConf
+
 
-        return cls(
-            api_id=conf["api"]["id"],
-            api_hash=conf["api"]["hash"],
+class VacuumConf(BaseModel):
+    window_delay_seconds: int | float
+    min_message_count: int
+    max_workers: int
 
-            clients_bots=conf["clients"]["bots"],
-            clients_userbots=conf["clients"]["userbots"],
 
-            heap_chat=conf["heap"]["chat"],
-            heap_page_fullness=conf["heap"]["page"]["fullness"],
+class RelationsConf(BaseModel):
+    chat: int
+    vacuum: VacuumConf
 
-            relations_chat=conf["relations"]["chat"],
-            relations_vacuum_window_delay_seconds=conf["relations"]["vacuum"]["window_delay_seconds"],
-            relations_vacuum_min_message_count=conf["relations"]["vacuum"]["min_message_count"],
 
-            buffer_chat=conf["buffer"]["chat"],
-            buffer_overflow_len=conf["buffer"]["overflow"]["len"],
-            buffer_overflow_timeout_seconds=conf["buffer"]["overflow"]["timeout_seconds"],
-            buffer_vacuum_window_delay_seconds=conf["buffer"]["vacuum"]["window_delay_seconds"],
-            buffer_vacuum_min_message_count=conf["buffer"]["vacuum"]["min_message_count"],
-        )
+class OverflowConf(BaseModel):
+    len: int
+    timeout_seconds: int | float
+
+
+class BufferConf(BaseModel):
+    chat: int
+    overflow: OverflowConf
+    vacuum: VacuumConf
+
+
+class Conf(BaseModel):
+    api: APIConf
+    clients: ClientsConf
+    horizon: HorizonConf
+    message_cache: MessageCacheConf
+    heap: HeapConf
+    relations: RelationsConf
+    buffer: BufferConf
+
+    @classmethod
+    def load(cls, path: Path) -> "Conf":
+        with path.open() as file:
+            data = yaml.safe_load(file)
+
+        conf = data["conf"]
+        return Conf(**conf)

src/tgdb/infrastructure/telethon/client_pool.py

@@ -72,7 +72,10 @@ def loaded_client_pool_from_farm_file(
     with farm_file_path.open() as farm_file:
         return TelegramClientPool(deque(
             TelegramClient(
-                StringSession(session_token), app_api_id, app_api_hash
+                StringSession(session_token),
+                app_api_id,
+                app_api_hash,
+                entity_cache=None,
             )
             for session_token in farm_file
             if session_token

src/tgdb/infrastructure/telethon/in_telegram_heap.py

@@ -1,6 +1,6 @@
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import cast
+from typing import ClassVar, cast
 
 from telethon.hints import TotalList
 
@@ -29,8 +29,19 @@ class InTelegramHeap:
     _encoded_tuple_max_len: int
     _message_map: LazyMessageMap
 
+    _page_len: ClassVar = 4000
+
+    @staticmethod
+    def encoded_tuple_max_len(page_fullness: float) -> int:
+        page_fullness = min(0, page_fullness)
+        page_fullness = max(page_fullness, 1)
+
+        return int(page_fullness * InTelegramHeap._page_len)
+
     def __post_init__(self) -> None:
-        assert_(self._encoded_tuple_max_len < 4000, ValueError)  # noqa: PLR2004
+        assert_(
+            self._encoded_tuple_max_len <= InTelegramHeap._page_len, ValueError
+        )
 
     def tuple_max_len(self) -> int:
         return self._encoded_tuple_max_len
@@ -82,24 +93,24 @@ async def tuples_with_attribute(
         )
 
     async def insert_idempotently(self, tuple: Tuple) -> None:
-        message_ = await self._message_map[tuple.tid]
+        message_ = await self._message_map[self._heap_id, tuple.tid]
 
         if message_ is not None:
             return
 
         new_message = await self._pool_to_insert().send_message(
             self._heap_id, HeapTupleEncoding.encoded_tuple(tuple)
         )
-        self._message_map[tuple.tid] = new_message
+        self._message_map[self._heap_id, tuple.tid] = new_message
 
     async def insert(self, tuple: Tuple) -> None:
         new_message = await self._pool_to_insert().send_message(
             self._heap_id, HeapTupleEncoding.encoded_tuple(tuple)
         )
-        self._message_map[tuple.tid] = new_message
+        self._message_map[self._heap_id, tuple.tid] = new_message
 
     async def update(self, tuple: Tuple) -> None:
-        message = await self._message_map[tuple.tid]
+        message = await self._message_map[self._heap_id, tuple.tid]
 
         if message is None:
             return
@@ -109,7 +120,7 @@ async def update(self, tuple: Tuple) -> None:
         )
 
     async def delete_tuple_with_tid(self, tid: TID) -> None:
-        message = await self._message_map[tid]
+        message = await self._message_map[self._heap_id, tid]
 
         if message is None:
             return

src/tgdb/infrastructure/telethon/lazy_message_map.py

@@ -9,13 +9,18 @@
 from tgdb.infrastructure.telethon.client_pool import TelegramClientPool
 
 
-type LazyMessageMap = LazyMap[TID, Message]
+type ChatID = int
+type LazyMessageMap = LazyMap[tuple[ChatID, TID], Message]
 
 
 def lazy_message_map(
-    chat_id: int, pool: TelegramClientPool, computed_map_max_len: int
+    pool: TelegramClientPool, cache_map_max_len: int
 ) -> LazyMessageMap:
-    async def tuple_message(tid: TID) -> ExternalValue[Message]:
+    async def tuple_message(
+        chat_id_and_tid: tuple[ChatID, TID]
+    ) -> ExternalValue[Message]:
+        chat_id, tid = chat_id_and_tid
+
         search = HeapTupleEncoding.id_of_encoded_tuple_with_tid(tid)
 
         messages = cast(TotalList, await pool().get_messages(
@@ -27,4 +32,4 @@ async def tuple_message(tid: TID) -> ExternalValue[Message]:
 
         return cast(Message, messages[0])
 
-    return LazyMap(computed_map_max_len, tuple_message)
+    return LazyMap(cache_map_max_len, tuple_message)