feat(infrastructure): add some adapters for telegram

Author: emptybutton

src/tgdb/application/output_commits_to_heap.py

@@ -1,19 +1,22 @@
 from dataclasses import dataclass
 
 from tgdb.application.ports.heap import Heap
+from tgdb.application.ports.message import TransactionCommitMessage
 from tgdb.application.ports.queque import Queque
-from tgdb.entities.transaction import (
-    TransactionCommit,
-    TransactionOkCommit,
-)
+from tgdb.entities.transaction import TransactionOkCommit
 
 
 @dataclass(frozen=True)
 class OutputCommitsToHeap:
     heap: Heap
-    output_commits: Queque[TransactionCommit]
+    output_commit_messages: Queque[TransactionCommitMessage]
 
     async def __call__(self) -> None:
-        async for commit in self.output_commits:
-            if isinstance(commit, TransactionOkCommit):
-                await self.heap.map(commit.effect)
+        async for message in self.output_commit_messages:
+            if not isinstance(message.commit, TransactionOkCommit):
+                return
+
+            if message.is_commit_duplicate:
+                await self.heap.map_as_duplicate(message.commit.effect)
+            else:
+                await self.heap.map(message.commit.effect)

src/tgdb/application/output_commits_to_log.py

@@ -1,27 +1,26 @@
 from dataclasses import dataclass
 
 from tgdb.application.ports.log import Log
+from tgdb.application.ports.message import TransactionCommitMessage
 from tgdb.application.ports.queque import Queque
 from tgdb.entities.operator import CommitOperator, StartOperator
-from tgdb.entities.transaction import (
-    TransactionCommit,
-    TransactionOkCommit,
-)
+from tgdb.entities.transaction import TransactionOkCommit
 
 
 @dataclass(frozen=True)
 class OutputCommitsToLog:
     log: Log
-    output_commits: Queque[TransactionCommit]
+    output_commit_messages: Queque[TransactionCommitMessage]
 
     async def __call__(self) -> None:
-        async for commit in self.output_commits:
-            if not isinstance(commit, TransactionOkCommit):
+        async for message in self.output_commit_messages:
+            if not isinstance(message.commit, TransactionOkCommit):
                 continue
 
-            start_operator = StartOperator(commit.transaction_id)
+            start_operator = StartOperator(message.commit.transaction_id)
             commit_operator = CommitOperator(
-                commit.transaction_id, commit.effect
+                message.commit.transaction_id, message.commit.effect
             )
 
-            await self.log.push_many((start_operator, commit_operator))
+            await self.log.push(start_operator)
+            await self.log.push(commit_operator)

src/tgdb/application/ports/heap.py

@@ -6,3 +6,6 @@
 class Heap(ABC):
     @abstractmethod
     async def map(self, effect: TransactionEffect, /) -> None: ...
+
+    @abstractmethod
+    async def map_as_duplicate(self, effect: TransactionEffect, /) -> None: ...

src/tgdb/application/ports/log.py

@@ -1,5 +1,4 @@
 from abc import ABC, abstractmethod
-from collections.abc import Sequence
 
 from tgdb.entities.logic_time import LogicTime
 from tgdb.entities.operator import AppliedOperator, Operator
@@ -10,12 +9,7 @@
 
 class Log(ABC):
     @abstractmethod
-    async def push_one(self, operator: Operator, /) -> AppliedOperator: ...
-
-    @abstractmethod
-    async def push_many(
-        self, operators: Sequence[Operator], /
-    ) -> Sequence[AppliedOperator]: ...
+    async def push(self, operator: Operator, /) -> AppliedOperator: ...
 
     @abstractmethod
     async def truncate(self, offset: LogOffset, /) -> None: ...

src/tgdb/application/ports/message.py

@@ -0,0 +1,9 @@
+from dataclasses import dataclass
+
+from tgdb.entities.transaction import TransactionCommit
+
+
+@dataclass(frozen=True)
+class TransactionCommitMessage:
+    commit: TransactionCommit
+    is_commit_duplicate: bool

src/tgdb/application/serialize_transactions.py

@@ -2,6 +2,7 @@
 
 from tgdb.application.ports.log import Log, LogOffset
 from tgdb.application.ports.log_iterator import LogIterator
+from tgdb.application.ports.message import TransactionCommitMessage
 from tgdb.application.ports.queque import Queque
 from tgdb.entities.logic_time import LogicTime
 from tgdb.entities.operator import AppliedOperator, Operator
@@ -17,7 +18,7 @@ class SerializeTransactions:
     log: Log
     log_iterator: LogIterator
     input_operators: Queque[Operator]
-    output_commits: Queque[TransactionCommit]
+    output_commit_messages: Queque[TransactionCommitMessage]
 
     async def __call__(
         self,
@@ -37,14 +38,19 @@ async def __call__(
         input_operators = aiter(self.input_operators)
 
         async for operator in self.log_iterator.finite():
-            await self._output_operator(operator, horizon)
+            await self._output_operator(operator, horizon, is_duplicate=True)
 
         async for operator in input_operators:
-            applied_operator = await self.log.push_one(operator)
-            await self._output_operator(applied_operator, horizon)
+            applied_operator = await self.log.push(operator)
+            await self._output_operator(
+                applied_operator, horizon, is_duplicate=False
+            )
 
     async def _output_operator(
-        self, operator: AppliedOperator, horizon: TransactionHorizon
+        self,
+        operator: AppliedOperator,
+        horizon: TransactionHorizon,
+        is_duplicate: bool,
     ) -> None:
         transaction_commit = horizon.add(operator)
 
@@ -54,10 +60,14 @@ async def _output_operator(
         )
 
         if transaction_commit:
-            await self.output_commits.push(transaction_commit)
+            await self.output_commit_messages.push(
+                TransactionCommitMessage(
+                    transaction_commit, is_commit_duplicate=is_duplicate
+                )
+            )
 
         if need_to_commit_offset:
-            await self.output_commits.sync()
+            await self.output_commit_messages.sync()
             await self.log_iterator.commit(offset_to_commit)
 
     def _safe_offset_to_commit(

src/tgdb/entities/mark.py

@@ -4,5 +4,5 @@
 
 @dataclass(frozen=True)
 class Mark:
-    mark_id: UUID
+    id: UUID
     key: str

src/tgdb/entities/row.py

@@ -49,16 +49,24 @@ def row(*attrs: RowAttribute, schema: Schema = "__undefined__") -> Row:
 class NewRow:
     row: Row
 
+    @property
+    def row_id(self) -> RowAttribute:
+        return self.row.id
+
 
 @dataclass(frozen=True)
 class MutatedRow:
     row: Row
     message: Message | None
 
+    @property
+    def row_id(self) -> RowAttribute:
+        return self.row.id
+
 
 @dataclass(frozen=True)
 class DeletedRow:
-    row: Row
+    row_id: RowAttribute
     message: Message | None
 
 

src/tgdb/infrastructure/adapters/heap.py

@@ -1,33 +1,66 @@
 from asyncio import gather
-from dataclasses import dataclass, field
-from itertools import groupby
+from dataclasses import dataclass
+
+from in_memory_db import InMemoryDb
 
 from tgdb.application.ports.heap import Heap
 from tgdb.entities.message import Message
-from tgdb.entities.row import DeletedRow, MutatedRow, NewRow, Row, RowEffect
+from tgdb.entities.row import (
+    DeletedRow,
+    MutatedRow,
+    NewRow,
+    Row,
+    RowEffect,
+)
 from tgdb.entities.transaction import TransactionEffect
-from tgdb.infrastructure.row_encoding import encoded_row
+from tgdb.infrastructure.heap_row_encoding import encoded_heap_row
 from tgdb.infrastructure.telethon.client_pool import TelegramClientPool
 from tgdb.infrastructure.telethon.lazy_message_map import LazyMessageMap
 from tgdb.infrastructure.telethon.mapping import message
 
 
 @dataclass(frozen=True, unsafe_hash=False)
 class InMemoryHeap(Heap):
-    _rows: set[Row] = field(default_factory=set)
+    _db: InMemoryDb[Row]
 
     async def map(self, effect: TransactionEffect) -> None:
-        for row_operator_effect in effect:
-            match row_operator_effect, row_operator_effect.row in self._rows:
-                case NewRow(row) | MutatedRow(row), _:
-                    self._rows.add(row)
-                case DeletedRow(row), True:
-                    self._rows.remove(row)
+        for row_effect in effect:
+            prevous_row = self._db.select_one(
+                lambda row: row.id == row_effect.row_id
+            )
+
+            match row_effect, prevous_row:
+                case DeletedRow(), Row():
+                    self._db.remove(prevous_row)
+
+                case NewRow(new_row), _:
+                    self._db.insert(new_row)
+
+                case MutatedRow(new_row), Row():
+                    self._db.remove(prevous_row)
+                    self._db.insert(new_row)
+
+                case _: ...
+
+    async def map_as_duplicate(self, effect: TransactionEffect) -> None:
+        for row_effect in effect:
+            prevous_row = self._db.select_one(
+                lambda row: row.id == row_effect.row_id
+            )
+
+            match row_effect, prevous_row:
+                case DeletedRow(row_id), Row():
+                    self._db.remove(prevous_row)
+
+                case NewRow(new_row) | MutatedRow(new_row), Row():
+                    self._db.remove(prevous_row)
+                    self._db.insert(new_row)
+
                 case _: ...
 
 
 @dataclass(frozen=True)
-class TelethonHeap(Heap):
+class InTelegramHeap(Heap):
     _pool_to_insert: TelegramClientPool
     _pool_to_select: TelegramClientPool
     _pool_to_edit: TelegramClientPool
@@ -36,34 +69,41 @@ class TelethonHeap(Heap):
     _message_map: LazyMessageMap
 
     async def map(self, transaction_effect: TransactionEffect) -> None:
-        row_effects_by_row_id = groupby(
-            transaction_effect, key=lambda it: it.row.id
-        )
-
         await gather(*(
-            self._map_row_effects(tuple(row_effects))
-            for _, row_effects in row_effects_by_row_id
+            self._map_row_effect(row_effect, is_duplicate=False)
+            for row_effect in transaction_effect
         ))
 
-    async def _map_row_effects(self, effect_part: TransactionEffect) -> None:
-        await gather(*map(self._map_row_effect, effect_part))
+    async def map_as_duplicate(
+        self, transaction_effect: TransactionEffect
+    ) -> None:
+        await gather(*(
+            self._map_row_effect(row_effect, is_duplicate=True)
+            for row_effect in transaction_effect
+        ))
 
-    async def _map_row_effect(self, row_effect: RowEffect) -> None:
+    async def _map_row_effect(
+        self, row_effect: RowEffect, *, is_duplicate: bool
+    ) -> None:
         match row_effect:
             case NewRow():
-                await self._insert(row_effect)
+                await self._insert(row_effect, is_duplicate)
             case MutatedRow():
                 await self._update(row_effect)
             case DeletedRow():
                 await self._delete(row_effect)
 
-    async def _insert(self, row_effect: NewRow) -> None:
-        telethon_message = await self._pool_to_insert().send_message(
-            self._heap_id, encoded_row(row_effect.row)
-        )
-        self._message_map[row_effect.row.schema, row_effect.row.id] = (
-            message(telethon_message)
+    async def _insert(self, row_effect: NewRow, is_duplicate: bool) -> None:
+        if is_duplicate:
+            message_ = await self._message_map[row_effect.row.id]
+
+            if message_ is not None:
+                return
+
+        tg_new_message = await self._pool_to_insert().send_message(
+            self._heap_id, encoded_heap_row(row_effect.row)
         )
+        self._message_map[row_effect.row.id] = message(tg_new_message)
 
     async def _update(self, row_effect: MutatedRow) -> None:
         message = await self._message(row_effect)
@@ -72,7 +112,7 @@ async def _update(self, row_effect: MutatedRow) -> None:
             return
 
         await self._pool_to_edit(message.author_id).edit_message(
-            self._heap_id, message.id, encoded_row(row_effect.row)
+            self._heap_id, message.id, encoded_heap_row(row_effect.row)
         )
 
     async def _delete(self, row_effect: DeletedRow) -> None:
@@ -91,4 +131,10 @@ async def _message(
         if row_effect.message is not None:
             return row_effect.message
 
-        return await self._message_map[row_effect.row.schema, row_effect.row.id]
+        match row_effect:
+            case MutatedRow():
+                row_id = row_effect.row.id
+            case DeletedRow():
+                row_id = row_effect.row_id
+
+        return await self._message_map[row_id]