feat: notify players about active payment (#4)

Author: emptybutton

src/ttt/application/player/complete_stars_purshase.py

@@ -3,28 +3,28 @@
 from ttt.application.common.ports.clock import Clock
 from ttt.application.common.ports.map import Map
 from ttt.application.common.ports.transaction import Transaction
+from ttt.application.player.ports.paid_stars_purchase_payment_inbox import (
+    PaidStarsPurchasePaymentInbox,
+)
 from ttt.application.player.ports.player_views import PlayerViews
 from ttt.application.player.ports.players import Players
-from ttt.application.player.ports.stars_purchase_payment_gateway import (
-    StarsPurchasePaymentGateway,
-)
 from ttt.entities.finance.payment.payment import PaymentAlreadyCompletedError
 from ttt.entities.tools.tracking import Tracking
 
 
 @dataclass(frozen=True, unsafe_hash=False)
 class CompleteStarsPurshase:
     clock: Clock
-    payment_gateway: StarsPurchasePaymentGateway
+    inbox: PaidStarsPurchasePaymentInbox
     players: Players
     transaction: Transaction
     map_: Map
     player_views: PlayerViews
 
     async def __call__(self) -> None:
-        current_datetime = await self.clock.current_datetime()
+        async for paid_payment in self.inbox.stream():
+            current_datetime = await self.clock.current_datetime()
 
-        async for paid_payment in self.payment_gateway.paid_payment_stream():
             async with self.transaction:
                 player = await self.players.player_with_id(
                     paid_payment.location.player_id,

src/ttt/application/player/dto/__init__.py

@@ -0,0 +1,12 @@
+from dataclasses import dataclass
+from uuid import UUID
+
+from ttt.entities.core.player.location import PlayerLocation
+from ttt.entities.finance.payment.success import PaymentSuccess
+
+
+@dataclass(frozen=True)
+class PaidStarsPurchasePayment:
+    purshase_id: UUID
+    location: PlayerLocation
+    success: PaymentSuccess

src/ttt/application/player/dto/common.py

@@ -0,0 +1,13 @@
+from abc import ABC, abstractmethod
+from collections.abc import AsyncIterable
+
+from ttt.application.player.dto.common import PaidStarsPurchasePayment
+
+
+class PaidStarsPurchasePaymentInbox(ABC):
+    @abstractmethod
+    async def push(self, payment: PaidStarsPurchasePayment) -> None: ...
+
+    @abstractmethod
+    def stream(self) -> AsyncIterable[PaidStarsPurchasePayment]:
+        ...

src/ttt/application/player/ports/paid_stars_purchase_payment_inbox.py

@@ -83,6 +83,11 @@ async def render_non_exchangeable_rubles_for_stars_view(
         self, location: PlayerLocation, /,
     ) -> None: ...
 
+    @abstractmethod
+    async def render_stars_purchase_will_be_completed_view(
+        self, location: PlayerLocation, /,
+    ) -> None: ...
+
     @abstractmethod
     async def render_completed_stars_purshase_view(
         self, player: Player, purshase_id: UUID, location: PlayerLocation, /,

src/ttt/application/player/ports/player_views.py

@@ -1,18 +1,9 @@
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable
-from dataclasses import dataclass
-from uuid import UUID
 
+from ttt.application.player.dto.common import PaidStarsPurchasePayment
 from ttt.entities.core.player.location import PlayerLocation
 from ttt.entities.core.player.stars_purchase import StarsPurchase
-from ttt.entities.finance.payment.success import PaymentSuccess
-
-
-@dataclass(frozen=True)
-class PaidStarsPurchasePayment:
-    purshase_id: UUID
-    location: PlayerLocation
-    success: PaymentSuccess
 
 
 class StarsPurchasePaymentGateway(ABC):

src/ttt/application/player/ports/stars_purchase_payment_gateway.py

@@ -0,0 +1,26 @@
+from dataclasses import dataclass
+
+from ttt.application.player.ports.paid_stars_purchase_payment_inbox import (
+    PaidStarsPurchasePaymentInbox,
+)
+from ttt.application.player.ports.player_views import PlayerViews
+from ttt.application.player.ports.stars_purchase_payment_gateway import (
+    StarsPurchasePaymentGateway,
+)
+
+
+@dataclass(frozen=True, unsafe_hash=False)
+class StartStarsPurshaseCompletion:
+    inbox: PaidStarsPurchasePaymentInbox
+    payment_gateway: StarsPurchasePaymentGateway
+    player_views: PlayerViews
+
+    async def __call__(self) -> None:
+        async for paid_payment in self.payment_gateway.paid_payment_stream():
+            await self.inbox.push(paid_payment)
+            await (
+                self.player_views
+                .render_stars_purchase_will_be_completed_view(
+                    paid_payment.location,
+                )
+            )

src/ttt/application/player/start_stars_purshase_completion.py

@@ -0,0 +1,22 @@
+from collections.abc import AsyncIterable
+from dataclasses import dataclass
+
+from ttt.application.player.dto.common import PaidStarsPurchasePayment
+from ttt.application.player.ports.paid_stars_purchase_payment_inbox import (
+    PaidStarsPurchasePaymentInbox,
+)
+from ttt.infrastructure.nats.paid_stars_purchase_payment_inbox import (
+    InNatsPaidStarsPurchasePaymentInbox as OriginalInNatsPaidStarsPurchasePaymentInbox,  # noqa: E501
+)
+
+
+@dataclass(frozen=True)
+class InNatsPaidStarsPurchasePaymentInbox(PaidStarsPurchasePaymentInbox):
+    _inbox: OriginalInNatsPaidStarsPurchasePaymentInbox
+
+    async def push(self, payment: PaidStarsPurchasePayment) -> None:
+        await self._inbox.push(payment)
+
+    async def stream(self) -> AsyncIterable[PaidStarsPurchasePayment]:
+        async for payment in self._inbox:
+            yield payment

src/ttt/infrastructure/adapters/paid_stars_purchase_payment_inbox.py

@@ -0,0 +1,29 @@
+from asyncio import Event
+from collections import deque
+from collections.abc import AsyncIterator
+from dataclasses import dataclass, field
+
+
+@dataclass(frozen=True, unsafe_hash=False)
+class Buffer[ValueT]:
+    _values: deque[ValueT] = field(default_factory=deque)
+    _has_values: Event = field(default_factory=Event, init=False)
+
+    def __post_init__(self) -> None:
+        if self._values:
+            self._has_values.set()
+
+    def __len__(self) -> int:
+        return len(self._values)
+
+    def add(self, value: ValueT) -> None:
+        self._values.append(value)
+        self._has_values.set()
+
+    async def stream(self) -> AsyncIterator[ValueT]:
+        while True:
+            await self._has_values.wait()
+            yield self._values.popleft()
+
+            if not self._values:
+                self._has_values.clear()

src/ttt/infrastructure/buffer.py

@@ -6,9 +6,7 @@
 from nats.js import JetStreamContext
 from pydantic import TypeAdapter
 
-from ttt.application.player.ports.stars_purchase_payment_gateway import (
-    PaidStarsPurchasePayment,
-)
+from ttt.application.player.dto.common import PaidStarsPurchasePayment
 from ttt.infrastructure.nats.messages import at_least_once_messages