feat(application): add a part of operations

Author: emptybutton

src/tgdb/application/commit_transaction.py

@@ -0,0 +1,97 @@
+from asyncio import gather
+from dataclasses import dataclass
+
+from tgdb.application.ports.buffer import Buffer
+from tgdb.application.ports.channel import Channel
+from tgdb.application.ports.clock import Clock
+from tgdb.application.ports.operator_encoding import (
+    DeletedTupleOperator,
+    MutatedTupleOperator,
+    NewTupleOperator,
+    Operator,
+    OperatorEncoding,
+)
+from tgdb.application.ports.relations import Relations
+from tgdb.application.ports.shared_horizon import SharedHorizon
+from tgdb.application.ports.uuids import UUIDs
+from tgdb.entities.horizon.claim import Claim
+from tgdb.entities.horizon.transaction import XID, Commit, PreparedCommit
+from tgdb.entities.relation.tuple_effect import (
+    DeletedTuple,
+    InvalidTuple,
+    MutatedTuple,
+    NewTuple,
+    deleted_tuple,
+    mutated_tuple,
+    new_tuple,
+)
+
+
+@dataclass(frozen=True)
+class CommitTransaction[EncodedOperatorsT]:
+    uuids: UUIDs
+    shared_horizon: SharedHorizon
+    clock: Clock
+    operator_encoding: OperatorEncoding[EncodedOperatorsT]
+    relations: Relations
+    channel: Channel
+    commit_buffer: Buffer[PreparedCommit]
+
+    async def __call__(
+        self, xid: XID, encoded_operators: EncodedOperatorsT
+    ) -> None:
+        """
+        :raises tgdb.entities.horizon.horizon.NoTransactionError:
+        :raises tgdb.entities.horizon.horizon.InvalidTransactionStateError:
+        :raises tgdb.entities.horizon.horizon.InvalidTupleError:
+        :raises tgdb.entities.horizon.horizon.InvalidEffectsError:
+        :raises tgdb.entities.horizon.transaction.ConflictError:
+        :raises tgdb.entities.horizon.transaction.NonSerializableWriteTransactionError:
+        """  # noqa: E501
+
+        operators = await self.operator_encoding.decoded(encoded_operators)
+
+        if operators is None:
+            effects = None
+        else:
+            effects = await gather(*map(self._effect, operators))
+
+        time = await self.clock
+
+        async with self.shared_horizon as horizon:
+            commit = horizon.commit_transaction(time, xid, effects)
+
+        if not isinstance(commit, PreparedCommit):
+            return
+
+        notification, _ = await gather(
+            self.channel.wait(commit.xid),
+            self.commit_buffer.add(commit),
+        )
+        if notification.error is not None:
+            raise notification.error from notification.error
+
+    async def _effect(
+        self, operator: Operator | None
+    ) -> NewTuple | MutatedTuple | DeletedTuple | Claim | InvalidTuple:
+        match operator:
+            case Claim():
+                return operator
+            case DeletedTupleOperator():
+                return deleted_tuple(operator.tid)
+            case None:
+                return InvalidTuple(None, None, None)
+            case _:
+                ...
+
+        if operator.relation_number is None:
+            relation = None
+        else:
+            relation = await self.relations.relation(operator.relation_number)
+
+        match operator:
+            case NewTupleOperator():
+                tid = await self.uuids.random_uuid()
+                return new_tuple(tid, operator.scalars, relation)
+            case MutatedTupleOperator():
+                return mutated_tuple(operator.tid, operator.scalars, relation)

src/tgdb/application/errors/__init__.py

@@ -2,31 +2,46 @@
 from dataclasses import dataclass
 
 from tgdb.application.ports.buffer import Buffer
-from tgdb.application.ports.notifying import Notifying
+from tgdb.application.ports.channel import Channel
+from tgdb.application.ports.clock import Clock
 from tgdb.application.ports.queque import Queque
 from tgdb.application.ports.shared_horizon import SharedHorizon
-from tgdb.entities.horizon.transaction import PreparedCommit
+from tgdb.entities.horizon.horizon import (
+    InvalidTransactionStateError,
+    NoTransactionError,
+)
+from tgdb.entities.horizon.transaction import XID, PreparedCommit
 
 
 @dataclass(frozen=True)
 class OutputCommits:
     commit_buffer: Buffer[PreparedCommit]
-    notifying: Notifying[Sequence[TransactionCommit]]
-    output_commits: Queque[Sequence[TransactionCommit]]
+    channel: Channel
+    output_commits: Queque[Sequence[PreparedCommit]]
     shared_horizon: SharedHorizon
+    clock: Clock
 
     async def __call__(self) -> None:
         async for prepared_commits in self.commit_buffer:
-            await self.notifying.publish(prepared_commits)
-
-            ok_prepared_commits = tuple(
-                commit for commit in prepared_commits
-                if isinstance(commit, TransactionCommit)
-            )
-
-            await self.output_commits.push(ok_prepared_commits)
+            await self.output_commits.push(prepared_commits)
             await self.output_commits.sync()
 
+            ok_commit_xids = list[XID]()
+            error_commit_map = dict[
+                XID, NoTransactionError | InvalidTransactionStateError
+            ]()
+
             async with self.shared_horizon as horizon:
-                for ok_prepared_commit in ok_prepared_commits:
-                    horizon.complete(ok_prepared_commit)
+                for prepared_commit in prepared_commits:
+                    time = await self.clock
+
+                    try:
+                        horizon.complete_commit(time, prepared_commit.xid)
+                    except (
+                        NoTransactionError, InvalidTransactionStateError
+                    ) as error:
+                        error_commit_map[prepared_commit.xid] = error
+                    else:
+                        ok_commit_xids.append(prepared_commit.xid)
+
+                await self.channel.publish(ok_commit_xids, error_commit_map)

src/tgdb/application/errors/common.py

@@ -0,0 +1,29 @@
+from abc import ABC, abstractmethod
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+
+from tgdb.entities.horizon.horizon import (
+    InvalidTransactionStateError,
+    NoTransactionError,
+)
+from tgdb.entities.horizon.transaction import XID
+
+
+@dataclass(frozen=True)
+class Notification:
+    error: NoTransactionError | InvalidTransactionStateError | None
+
+
+class Channel(ABC):
+    @abstractmethod
+    async def publish(
+        self,
+        ok_commit_xids: Sequence[XID],
+        error_commit_map: Mapping[
+            XID, NoTransactionError | InvalidTransactionStateError
+        ],
+        /,
+    ) -> None: ...
+
+    @abstractmethod
+    async def wait(self, xid: XID) -> Notification: ...

src/tgdb/application/input_operator.py

@@ -1,6 +1,7 @@
+from abc import ABC
 from collections.abc import Awaitable
 
 from tgdb.entities.time.logic_time import LogicTime
 
 
-class Clock(Awaitable[LogicTime]): ...
+class Clock(ABC, Awaitable[LogicTime]): ...