server/subscriptions: implement retry and precondition error handling…

… for benefit grants

server/polar/subscription/email_templates/benefits/custom/precondition_failed.html

@@ -0,0 +1,6 @@
+{% extends "base.html" %}
+
+{% block body %}
+<h1>Uh oh 😫</h1>
+<p>We had trouble granting you access to the benefit {{ subscription_benefit.description }}.</p>
+{% endblock %}

server/polar/subscription/service/benefits/__init__.py

@@ -7,8 +7,8 @@
 from ...schemas import SubscriptionBenefitUpdate
 from .base import (
     SB,
-    SubscriptionBenefitGrantError,
-    SubscriptionBenefitRevokeError,
+    SubscriptionBenefitPreconditionError,
+    SubscriptionBenefitRetriableError,
     SubscriptionBenefitServiceError,
     SubscriptionBenefitServiceProtocol,
 )
@@ -30,8 +30,8 @@ def get_subscription_benefit_service(
 
 __all__ = [
     "SubscriptionBenefitServiceProtocol",
-    "SubscriptionBenefitGrantError",
-    "SubscriptionBenefitRevokeError",
+    "SubscriptionBenefitPreconditionError",
+    "SubscriptionBenefitRetriableError",
     "SubscriptionBenefitServiceError",
     "get_subscription_benefit_service",
 ]

server/polar/subscription/service/benefits/base.py

@@ -1,7 +1,7 @@
-from typing import Protocol, TypeVar
+from typing import Any, Protocol, TypeVar
 
 from polar.exceptions import PolarError
-from polar.models import SubscriptionBenefit
+from polar.models import Subscription, SubscriptionBenefit
 from polar.postgres import AsyncSession
 
 from ...schemas import SubscriptionBenefitUpdate
@@ -11,29 +11,128 @@ class SubscriptionBenefitServiceError(PolarError):
     ...
 
 
-class SubscriptionBenefitGrantError(SubscriptionBenefitServiceError):
-    ...
-
-
-class SubscriptionBenefitRevokeError(SubscriptionBenefitServiceError):
-    ...
+class SubscriptionBenefitRetriableError(SubscriptionBenefitServiceError):
+    """
+    A retriable error occured while granting or revoking the benefit.
+    """
+
+    defer_seconds: int
+    "Number of seconds to wait before retrying."
+
+    def __init__(self, defer_seconds: int) -> None:
+        self.defer_seconds = defer_seconds
+        message = f"An error occured. We'll retry in {defer_seconds} seconds."
+        super().__init__(message)
+
+
+class SubscriptionBenefitPreconditionError(SubscriptionBenefitServiceError):
+    """
+    Some conditions are missing to grant the benefit.
+
+    It accepts an email subject and body templates. When set, an email will
+    be sent to the backer to explain them what happened. It'll be generated with
+    the following context:
+
+    ```py
+    class Context:
+        subscription: Subscription
+        subscription_tier: SubscriptionTier
+        subscription_benefit: SubscriptionBenefit
+        user: User
+    ```
+
+    An additional context dictionary can also be passed.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        email_subject: str | None = None,
+        email_body_template: str | None = None,
+        email_extra_context: dict[str, Any] | None = None,
+    ) -> None:
+        """
+        Args:
+            message: The plain error message
+            email_subject: Template string for the email subject
+            we'll send to the backer.
+            email_body_template: Path to the email template body
+            we'll send to the backer.
+            It's expected to be under `subscription/email_templates` directory.
+        """
+        self.email_subject = email_subject
+        self.email_body_template = email_body_template
+        self.email_extra_context = email_extra_context or {}
+        super().__init__(message)
 
 
 SB = TypeVar("SB", bound=SubscriptionBenefit, contravariant=True)
 SBU = TypeVar("SBU", bound=SubscriptionBenefitUpdate, contravariant=True)
 
 
 class SubscriptionBenefitServiceProtocol(Protocol[SB, SBU]):
+    """
+    Protocol that should be implemented by each benefit type service.
+
+    It allows to implement very customizable and specific logic to fulfill the benefit.
+    """
+
     session: AsyncSession
 
     def __init__(self, session: AsyncSession) -> None:
         self.session = session
 
-    async def grant(self, benefit: SB) -> None:
+    async def grant(
+        self, benefit: SB, subscription: Subscription, *, attempt: int = 1
+    ) -> None:
+        """
+        Executes the logic to grant a benefit to a backer.
+
+        Args:
+            benefit: The SubscriptionBenefit to grant.
+            subscription: The Subscription we should grant this benefit to.
+            Use it to access the underlying backer user.
+            attempt: Number of times we attempted to grant the benefit.
+            Useful for the worker to implement retry logic.
+
+        Raises:
+            SubscriptionBenefitRetriableError: An temporary error occured,
+            we should be able to retry later.
+            SubscriptionBenefitPreconditionError: Some conditions are missing
+            to grant the benefit.
+        """
         ...
 
-    async def revoke(self, benefit: SB) -> None:
+    async def revoke(
+        self, benefit: SB, subscription: Subscription, *, attempt: int = 1
+    ) -> None:
+        """
+        Executes the logic to revoke a benefit from a backer.
+
+        Args:
+            benefit: The SubscriptionBenefit to revoke.
+            subscription: The Subscription we should revoke this benefit from.
+            Use it to access the underlying backer user.
+            attempt: Number of times we attempted to revoke the benefit.
+            Useful for the worker to implement retry logic.
+
+        Raises:
+            SubscriptionBenefitRetriableError: An temporary error occured,
+            we should be able to retry later.
+        """
         ...
 
     async def requires_update(self, benefit: SB, update: SBU) -> bool:
+        """
+        Determines if a benefit update requires to trigger the granting logic again.
+
+        This method is called whenever a benefit is updated. If it returns `True`, the
+        granting logic will be re-executed again for all the backers.
+
+        Args:
+            benefit: The updated SubscriptionBenefit.
+            update: The SubscriptionBenefitUpdate schema.
+            Use it to check which fields have been updated.
+        """
         ...

server/polar/subscription/service/benefits/custom.py

@@ -1,3 +1,4 @@
+from polar.models import Subscription
 from polar.models.subscription_benefit import SubscriptionBenefitCustom
 
 from ...schemas import SubscriptionBenefitCustomUpdate
@@ -9,10 +10,22 @@ class SubscriptionBenefitCustomService(
         SubscriptionBenefitCustom, SubscriptionBenefitCustomUpdate
     ]
 ):
-    async def grant(self, benefit: SubscriptionBenefitCustom) -> None:
+    async def grant(
+        self,
+        benefit: SubscriptionBenefitCustom,
+        subscription: Subscription,
+        *,
+        attempt: int = 1,
+    ) -> None:
         return
 
-    async def revoke(self, benefit: SubscriptionBenefitCustom) -> None:
+    async def revoke(
+        self,
+        benefit: SubscriptionBenefitCustom,
+        subscription: Subscription,
+        *,
+        attempt: int = 1,
+    ) -> None:
         return
 
     async def requires_update(

server/polar/subscription/service/subscription_benefit_grant.py

@@ -1,14 +1,23 @@
 from collections.abc import Sequence
 
+import structlog
 from sqlalchemy import select
 
+from polar.email.renderer import get_email_renderer
+from polar.email.sender import get_email_sender
 from polar.kit.services import ResourceServiceReader
+from polar.logging import Logger
 from polar.models import Subscription, SubscriptionBenefit, SubscriptionBenefitGrant
 from polar.postgres import AsyncSession
 from polar.worker import enqueue_job
 
 from ..schemas import SubscriptionBenefitUpdate
-from .benefits import get_subscription_benefit_service
+from .benefits import (
+    SubscriptionBenefitPreconditionError,
+    get_subscription_benefit_service,
+)
+
+log: Logger = structlog.get_logger()
 
 
 class SubscriptionBenefitGrantService(ResourceServiceReader[SubscriptionBenefitGrant]):
@@ -17,6 +26,8 @@ async def grant_benefit(
         session: AsyncSession,
         subscription: Subscription,
         subscription_benefit: SubscriptionBenefit,
+        *,
+        attempt: int = 1,
     ) -> SubscriptionBenefitGrant:
         grant = await self._get_by_subscription_and_benefit(
             session, subscription, subscription_benefit
@@ -32,9 +43,17 @@ async def grant_benefit(
         benefit_service = get_subscription_benefit_service(
             subscription_benefit.type, session
         )
-        await benefit_service.grant(subscription_benefit)
-
-        grant.set_granted()
+        try:
+            await benefit_service.grant(
+                subscription_benefit, subscription, attempt=attempt
+            )
+        except SubscriptionBenefitPreconditionError as e:
+            await self.handle_precondition_error(
+                session, e, subscription, subscription_benefit
+            )
+            grant.granted_at = None
+        else:
+            grant.set_granted()
 
         session.add(grant)
         await session.commit()
@@ -46,6 +65,8 @@ async def revoke_benefit(
         session: AsyncSession,
         subscription: Subscription,
         subscription_benefit: SubscriptionBenefit,
+        *,
+        attempt: int = 1,
     ) -> SubscriptionBenefitGrant:
         grant = await self._get_by_subscription_and_benefit(
             session, subscription, subscription_benefit
@@ -61,7 +82,9 @@ async def revoke_benefit(
         benefit_service = get_subscription_benefit_service(
             subscription_benefit.type, session
         )
-        await benefit_service.revoke(subscription_benefit)
+        await benefit_service.revoke(
+            subscription_benefit, subscription, attempt=attempt
+        )
 
         grant.set_revoked()
 
@@ -95,20 +118,31 @@ async def update_benefit_grant(
         self,
         session: AsyncSession,
         grant: SubscriptionBenefitGrant,
+        *,
+        attempt: int = 1,
     ) -> SubscriptionBenefitGrant:
         # Don't update revoked benefits
         if grant.is_revoked:
             return grant
 
-        await session.refresh(grant, {"subscription_benefit"})
+        await session.refresh(grant, {"subscription", "subscription_benefit"})
+        subscription = grant.subscription
         subscription_benefit = grant.subscription_benefit
 
         benefit_service = get_subscription_benefit_service(
             subscription_benefit.type, session
         )
-        await benefit_service.grant(subscription_benefit)
-
-        grant.set_granted()
+        try:
+            await benefit_service.grant(
+                subscription_benefit, subscription, attempt=attempt
+            )
+        except SubscriptionBenefitPreconditionError as e:
+            await self.handle_precondition_error(
+                session, e, subscription, subscription_benefit
+            )
+            grant.granted_at = None
+        else:
+            grant.set_granted()
 
         session.add(grant)
         await session.commit()
@@ -129,18 +163,23 @@ async def delete_benefit_grant(
         self,
         session: AsyncSession,
         grant: SubscriptionBenefitGrant,
+        *,
+        attempt: int = 1,
     ) -> SubscriptionBenefitGrant:
         # Already revoked, nothing to do
         if grant.is_revoked:
             return grant
 
-        await session.refresh(grant, {"subscription_benefit"})
+        await session.refresh(grant, {"subscription", "subscription_benefit"})
+        subscription = grant.subscription
         subscription_benefit = grant.subscription_benefit
 
         benefit_service = get_subscription_benefit_service(
             subscription_benefit.type, session
         )
-        await benefit_service.revoke(subscription_benefit)
+        await benefit_service.revoke(
+            subscription_benefit, subscription, attempt=attempt
+        )
 
         grant.set_revoked()
 
@@ -149,6 +188,41 @@ async def delete_benefit_grant(
 
         return grant
 
+    async def handle_precondition_error(
+        self,
+        session: AsyncSession,
+        error: SubscriptionBenefitPreconditionError,
+        subscription: Subscription,
+        subscription_benefit: SubscriptionBenefit,
+    ) -> None:
+        if error.email_subject is None or error.email_body_template is None:
+            log.warning(
+                "A precondition error was raised but the user was not notified. "
+                "We probably should implement an email for this error.",
+                subscription_id=str(subscription.id),
+                subscription_benefit_id=str(subscription_benefit.id),
+            )
+            return
+
+        email_renderer = get_email_renderer({"subscription": "polar.subscription"})
+        email_sender = get_email_sender()
+
+        await session.refresh(subscription, {"user", "subscription_tier"})
+
+        subject, body = email_renderer.render_from_template(
+            error.email_subject,
+            f"subscription/{error.email_body_template}",
+            {
+                "subscription": subscription,
+                "subscription_tier": subscription.subscription_tier,
+                "subscription_benefit": subscription_benefit,
+                "user": subscription.user,
+                **error.email_extra_context,
+            },
+        )
+
+        email_sender.send_to_user(subscription.user.email, subject, body)
+
     async def _get_by_subscription_and_benefit(
         self,
         session: AsyncSession,