[feat] Stripe subscription (#3369)

lnbits/fiat/base.py

@@ -4,6 +4,8 @@ from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator, Coroutine
 from typing import TYPE_CHECKING, Any, NamedTuple
 
+from pydantic import BaseModel, Field
+
 if TYPE_CHECKING:
     pass
 
@@ -76,6 +78,54 @@ class FiatPaymentStatus(NamedTuple):
         return "pending"
 
 
+class FiatSubscriptionPaymentOptions(BaseModel):
+
+    memo: str | None = Field(
+        default=None,
+        description="Payments created by the recurring subscription"
+        " will have this memo.",
+    )
+    wallet_id: str | None = Field(
+        default=None,
+        description="Payments created by the recurring subscription"
+        " will be made to this wallet.",
+    )
+    subscription_request_id: str | None = Field(
+        default=None,
+        description="Unique ID that can be used to identify the subscription request."
+        "If not provided, one will be generated.",
+    )
+    tag: str | None = Field(
+        default=None,
+        description="Payments created by the recurring subscription"
+        " will have this tag. Admin only.",
+    )
+    extra: dict[str, Any] | None = Field(
+        default=None,
+        description="Payments created by the recurring subscription"
+        " will merge this extra data to the payment extra. Admin only.",
+    )
+
+    success_url: str | None = Field(
+        default="https://my.lnbits.com",
+        description="The URL to redirect the user to after the"
+        " subscription is successfully created.",
+    )
+
+
+class CreateFiatSubscription(BaseModel):
+    subscription_id: str
+    quantity: int
+    payment_options: FiatSubscriptionPaymentOptions
+
+
+class FiatSubscriptionResponse(BaseModel):
+    ok: bool = True
+    subscription_request_id: str | None = None
+    checkout_session_url: str | None = None
+    error_message: str | None = None
+
+
 class FiatPaymentSuccessStatus(FiatPaymentStatus):
     paid = True
 
@@ -111,6 +161,32 @@ class FiatProvider(ABC):
     ) -> Coroutine[None, None, FiatInvoiceResponse]:
         pass
 
+    @abstractmethod
+    def create_subscription(
+        self,
+        subscription_id: str,
+        quantity: int,
+        payment_options: FiatSubscriptionPaymentOptions,
+        **kwargs,
+    ) -> Coroutine[None, None, FiatSubscriptionResponse]:
+        pass
+
+    @abstractmethod
+    def cancel_subscription(
+        self,
+        subscription_id: str,
+        correlation_id: str,
+        **kwargs,
+    ) -> Coroutine[None, None, FiatSubscriptionResponse]:
+        """
+        Cancel a subscription.
+        Args:
+            subscription_id: The ID of the subscription to cancel.
+            correlation_id: An identifier used to verify that the subscription belongs
+                            to the user that made the request. Usually the wallet ID.
+        """
+        pass
+
     @abstractmethod
     def pay_invoice(
         self,