ref: make infinite interactors not infinite (#61)

Author: emptybutton

deploy/dev/docker-compose.yaml

@@ -39,6 +39,8 @@ services:
       TTT_MATCHMAKING_MAX_WORKERS: 4
       TTT_MATCHMAKING_WORKER_MAX_USERS: 100
       TTT_MATCHMAKING_WORKER_CREATION_INTERVAL_SECONDS: 0.5
+
+      TTT_AUTO_CANCEL_INVITATIONS_TO_GAME_INTERVAL_SECONDS: 1
     secrets:
       - secrets
     command: ttt-dev

deploy/prod/docker-compose.yaml

@@ -35,6 +35,12 @@ services:
       TTT_NATS_URL: nats://${NATS_TOKEN}@nats:4222
 
       TTT_GEMINI_URL: ${GEMINI_URL}
+
+      TTT_MATCHMAKING_MAX_WORKERS: 4
+      TTT_MATCHMAKING_WORKER_MAX_USERS: 100
+      TTT_MATCHMAKING_WORKER_CREATION_INTERVAL_SECONDS: 0.5
+
+      TTT_AUTO_CANCEL_INVITATIONS_TO_GAME_INTERVAL_SECONDS: 1
     secrets:
       - secrets
     networks:

src/ttt/application/stars_purchase/complete_stars_purchase_payment.py

@@ -32,7 +32,7 @@ class CompleteStarsPurchasePayment:
     stars_purchases: StarsPurchases
 
     async def __call__(self) -> None:
-        async for paid_payment in self.inbox.stream():
+        async for paid_payment in self.inbox:
             current_datetime = await self.clock.current_datetime()
 
             async with self.transaction:

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

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterator
 
 from ttt.application.stars_purchase.dto.common import PaidStarsPurchasePayment
 
@@ -9,4 +9,4 @@ class PaidStarsPurchasePaymentInbox(ABC):
     async def push(self, payment: PaidStarsPurchasePayment) -> None: ...
 
     @abstractmethod
-    def stream(self) -> AsyncIterable[PaidStarsPurchasePayment]: ...
+    def __aiter__(self) -> AsyncIterator[PaidStarsPurchasePayment]: ...

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

@@ -1,8 +1,6 @@
 from abc import ABC, abstractmethod
-from collections.abc import AsyncIterable
 from uuid import UUID
 
-from ttt.application.stars_purchase.dto.common import PaidStarsPurchasePayment
 from ttt.entities.core.stars_purchase.stars_purchase import StarsPurchase
 
 
@@ -18,8 +16,3 @@ async def stop_payment_due_to_dublicate(self, payment_id: UUID) -> None: ...
 
     @abstractmethod
     async def stop_payment_due_to_error(self, payment_id: UUID) -> None: ...
-
-    @abstractmethod
-    def paid_payment_stream(
-        self,
-    ) -> AsyncIterable[PaidStarsPurchasePayment]: ...

src/ttt/application/stars_purchase/start_stars_purchase_payment_completion.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 
+from ttt.application.stars_purchase.dto.common import PaidStarsPurchasePayment
 from ttt.application.stars_purchase.ports.paid_stars_purchase_payment_inbox import (  # noqa: E501
     PaidStarsPurchasePaymentInbox,
 )
@@ -21,12 +22,11 @@ class StartStarsPurchasePaymentCompletion:
     views: StarsPurchaseViews
     log: StarsPurchaseLog
 
-    async def __call__(self) -> None:
-        async for paid_payment in self.payment_gateway.paid_payment_stream():
-            await self.inbox.push(paid_payment)
-            await self.views.stars_purchase_will_be_completed_view(
-                paid_payment.user_id,
-            )
-            await self.log.stars_purchase_payment_completion_started(
-                paid_payment,
-            )
+    async def __call__(self, paid_payment: PaidStarsPurchasePayment) -> None:
+        await self.inbox.push(paid_payment)
+        await self.log.stars_purchase_payment_completion_started(
+            paid_payment,
+        )
+        await self.views.stars_purchase_will_be_completed_view(
+            paid_payment.user_id,
+        )

src/ttt/infrastructure/adapters/paid_stars_purchase_payment_inbox.py

@@ -1,4 +1,4 @@
-from collections.abc import AsyncIterable
+from collections.abc import AsyncIterator
 from dataclasses import dataclass
 
 from ttt.application.stars_purchase.dto.common import PaidStarsPurchasePayment
@@ -17,6 +17,6 @@ class InNatsPaidStarsPurchasePaymentInbox(PaidStarsPurchasePaymentInbox):
     async def push(self, payment: PaidStarsPurchasePayment) -> None:
         await self._inbox.push(payment)
 
-    async def stream(self) -> AsyncIterable[PaidStarsPurchasePayment]:
+    async def __aiter__(self) -> AsyncIterator[PaidStarsPurchasePayment]:
         async for payment in self._inbox:
             yield payment

src/ttt/infrastructure/background_tasks.py

@@ -7,21 +7,20 @@
 
 async def at_least_once_messages(
     subscription: JetStreamContext.PullSubscription,
-    batch: int = 1,
+    max_len: int = 1,
     timeout: float | None = 5,  # noqa: ASYNC109
     heartbeat: float | None = None,
 ) -> AsyncIterator[Msg]:
-    while True:
-        try:
-            messages = await subscription.fetch(batch, timeout, heartbeat)
-        except NatsTimeoutError:
-            continue
+    try:
+        messages = await subscription.fetch(max_len, timeout, heartbeat)
+    except NatsTimeoutError:
+        return
 
-        for message in messages:
-            try:
-                yield message
-            except BaseException as error:
-                await message.nak()
-                raise error from error
-            else:
-                await message.ack()
+    for message in messages:
+        try:
+            yield message
+        except BaseException as error:
+            await message.nak()
+            raise error from error
+        else:
+            await message.ack()