Coverage for functions \ flipdare \ service \ payments \ _payment_link_handler.py: 53%
125 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
3#
4# This file is part of Flipdare's proprietary software and contains
5# confidential and copyrighted material. Unauthorised copying,
6# modification, distribution, or use of this file is strictly
7# prohibited without prior written permission from Flipdare Pty Ltd.
8#
9# This software includes third-party components licensed under MIT,
10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
11#
13from __future__ import annotations
15from typing import TYPE_CHECKING
16from stripe import AccountLink as StripeAccountLink
17from typing import Any
18from firebase_functions import https_fn
19from flipdare.app_log import LOG
20from flipdare.constants import IS_DEBUG
21from flipdare.message.error_message import ErrorMessage
22from flipdare.error import (
23 AppError,
24 StripeErrorContext,
25)
26from flipdare.generated.shared.app_payment_error_code import AppPaymentErrorCode
27from flipdare.service._error_mixin import ErrorMixin
28from flipdare.generated import (
29 ErrorSchema,
30 StripeAccountType,
31 StripeLinkResponseSchema,
32)
33from flipdare.payments.app_stripe_proxy import AppStripeProxy
34from flipdare.payments.app_stripe_config import (
35 StripeLinkInfo,
36)
37from flipdare.request import (
38 StripeRefreshAccountRequestAdapter,
39)
40from flipdare.service._service_provider import ServiceProvider
41from flipdare.service._user_mixin import UserMixin
42from flipdare.service.payments._base_payment_handler import BasePaymentHandler
43from flipdare.payments.core.stripe_guard import StripeGuard
44from flipdare.wrapper.user_wrapper import UserWrapper
46if TYPE_CHECKING:
47 from flipdare.manager.service_manager import ServiceManager
48 from flipdare.manager.backend_manager import BackendManager
49 from flipdare.manager.db_manager import DbManager
51__all__ = ["PaymentLinkHandler"]
54class PaymentLinkHandler(BasePaymentHandler, ErrorMixin, UserMixin, ServiceProvider):
55 def __init__(
56 self,
57 proxy: AppStripeProxy,
58 db_manager: DbManager | None = None,
59 backend_manager: BackendManager | None = None,
60 service_manager: ServiceManager | None = None,
61 ) -> None:
62 self.proxy = proxy
63 super().__init__(
64 db_manager=db_manager,
65 backend_manager=backend_manager,
66 service_manager=service_manager,
67 )
69 # ----------------------------------------------------------------------------------------------
70 # CALLABLE REQUESTS
71 # ----------------------------------------------------------------------------------------------
73 def callable_create_onboard_link(
74 self,
75 req: https_fn.CallableRequest[Any],
76 ) -> StripeLinkResponseSchema | ErrorSchema:
77 request_adapter: StripeRefreshAccountRequestAdapter | None = None
78 endpoint = req.raw_request.endpoint or "(internal)create_onboard_link"
79 try:
80 request_adapter = StripeRefreshAccountRequestAdapter.from_callable(req)
81 request_adapter.validate()
82 except AppError as e:
83 cause = f"Onboard link request validation error: {e}\n\tRequest: {req!s}"
84 LOG().error(cause)
85 error = StripeErrorContext.from_code(
86 endpoint=endpoint,
87 error_code=AppPaymentErrorCode.INVALID_REQUEST,
88 cause=cause,
89 error=e,
90 )
91 return error.to_dict()
93 uid = request_adapter.uid
94 try:
95 user = self.get_user_by_id(endpoint=request_adapter.endpoint, uid=uid)
96 except Exception as e:
97 msg = f"Failed to get user info for uid={uid}: {e}"
98 LOG().error(msg)
99 error = StripeErrorContext.from_code(
100 endpoint=endpoint,
101 error_code=AppPaymentErrorCode.INVALID_USER,
102 cause=msg,
103 error=e,
104 )
105 return error.to_dict()
107 settings = user.stripe_settings
108 if not StripeGuard.is_account(settings):
109 msg = f"Unable to refresh URL for user {user.doc_id}: No stripe settings.."
110 LOG().error(msg)
111 error = StripeErrorContext.from_code(
112 endpoint=endpoint,
113 error_code=AppPaymentErrorCode.INVALID_SETTINGS,
114 cause=msg,
115 )
116 return error.to_dict()
118 account_id = request_adapter.stripe_account_id
119 account_type = request_adapter.account_type
120 country_code = settings.country_code
121 currency_code = settings.currency_code
123 # we dont worry about changes, the db is updated when the link is created..
124 settings.update(
125 account_id=account_id,
126 country_code=country_code,
127 currency_code=currency_code,
128 account_type=account_type,
129 )
131 try:
132 return self.create_onboard_link(
133 endpoint=request_adapter.endpoint,
134 user=user,
135 # we want to force a new link to be created when refreshing,
136 # since the old one is no longer valid
137 overwrite=True,
138 )
139 except Exception as e:
140 msg = f"Failed to create onboard link for user {user.doc_id}\n\tError:{e}"
141 LOG().error(msg)
142 error = StripeErrorContext.from_code(
143 endpoint=endpoint,
144 error_code=AppPaymentErrorCode.LINK_CREATE_FAILED,
145 cause=msg,
146 error=e,
147 )
148 return error.to_dict()
150 # ----------------------------------------------------------------------------------------------
151 # LINK REQUESTS
152 # ----------------------------------------------------------------------------------------------
154 def create_onboard_link(
155 self,
156 endpoint: str,
157 user: UserWrapper,
158 overwrite: bool = False,
159 ) -> StripeLinkResponseSchema:
160 account_link: StripeLinkResponseSchema | None = None
161 settings = user.stripe_settings
162 if not StripeGuard.is_account(settings):
163 msg = f"User {user.doc_id} does not have a valid stripe account in settings.."
164 LOG().error(msg)
165 error = StripeErrorContext.from_code(
166 endpoint=endpoint,
167 error_code=AppPaymentErrorCode.INVALID_SETTINGS,
168 cause=msg,
169 )
170 raise AppError.from_context(error)
172 account_id = settings.account_id
173 if not overwrite:
174 account_link = self.get_stored_onboard_link(user)
175 if account_link is not None:
176 msg = f"Existing account link for {account_id} (force={overwrite}): {account_link}"
177 LOG().debug(msg)
178 return account_link
180 onboard_link = self._create_new_onboard_link(
181 endpoint=endpoint,
182 uid=user.doc_id,
183 account_id=account_id,
184 account_type=settings.account_type,
185 )
187 user.update_stripe_account(
188 account_link=onboard_link["url"],
189 account_link_expires=onboard_link["expires_at"],
190 # existing settings, but required to avoid any promotion/demotion issues.
191 account_id=account_id,
192 account_type=settings.account_type,
193 currency_code=settings.currency_code,
194 )
196 try:
197 self.update_user(
198 endpoint=endpoint,
199 user=user,
200 on_error_msg=ErrorMessage.STR_ACC_UPDATE_FAILED,
201 )
203 return StripeLinkResponseSchema(
204 {
205 "url": onboard_link["url"],
206 "expires_at": onboard_link["expires_at"],
207 "account_id": account_id,
208 }
209 )
210 except Exception as e:
211 msg = f"Failed to save account link for {account_id} in db\n\tError:{e}"
212 LOG().error(msg)
213 error = StripeErrorContext.from_code(
214 endpoint=endpoint,
215 error_code=AppPaymentErrorCode.LINK_CREATE_FAILED,
216 cause=msg,
217 error=e,
218 )
219 raise AppError.from_context(error) from e
221 def get_stored_onboard_link(
222 self,
223 user: UserWrapper,
224 ) -> StripeLinkResponseSchema | None:
225 stripe_settings = user.stripe_settings
226 if not StripeGuard.is_account(stripe_settings):
227 msg = f"User {user.doc_id} does not have valid stripe account settings.."
228 LOG().error(msg)
229 return None
231 is_active = stripe_settings.is_link_active
232 if not is_active:
233 if IS_DEBUG:
234 msg = (
235 f"Existing link {user.doc_id} "
236 f"({stripe_settings.account_link_expires}) expired or inactive."
237 )
238 LOG().debug(msg)
240 return None
242 account_link = stripe_settings.account_link
243 account_id = stripe_settings.account_id
244 account_link_expires = stripe_settings.account_link_expires
246 if account_link is None or account_link_expires is None:
247 msg = (
248 f"Missing fields for link for {user.doc_id}: "
249 f"account_link={account_link}, account_id={account_id}, "
250 f"account_link_expires={account_link_expires}"
251 )
252 LOG().error(msg)
253 return None
255 if IS_DEBUG:
256 msg = (
257 f"Existing account link for user {user.doc_id}/{account_id} is active: "
258 f"{account_link}, expires at {account_link_expires}"
259 )
260 LOG().debug(msg)
262 link = StripeLinkResponseSchema(
263 url=account_link,
264 expires_at=account_link_expires,
265 account_id=account_id,
266 )
268 if IS_DEBUG:
269 msg = f"Existing link still active for {user.doc_id} : {link}"
270 LOG().debug(msg)
272 return link
274 # ----------------------------------------------------------------------------------------------
275 # INTERNAL: LINKs
276 # ----------------------------------------------------------------------------------------------
278 def _create_new_onboard_link(
279 self,
280 endpoint: str,
281 uid: str,
282 account_id: str,
283 account_type: StripeAccountType,
284 ) -> StripeLinkResponseSchema:
285 account_link: StripeAccountLink | None = None
287 try:
288 info = StripeLinkInfo(
289 uid=uid,
290 stripe_account_id=account_id,
291 account_type=account_type,
292 )
293 account_link = self.proxy.create_onboard_link(endpoint, info)
295 except Exception as e:
296 msg = f"Failed to create account link for {account_id} for user {uid}\n\tError:{e}"
297 LOG().error(msg)
298 error = StripeErrorContext.from_code(
299 endpoint=endpoint,
300 error_code=AppPaymentErrorCode.LINK_CREATE_FAILED,
301 cause=msg,
302 error=e,
303 )
304 raise AppError.from_context(error) from e
306 return StripeLinkResponseSchema(
307 url=account_link.url,
308 expires_at=account_link.expires_at,
309 account_id=account_id,
310 )