✨ Introduce wallet exchange 🗃️

api/specs/web-server/_projects_wallet.py

@@ -8,15 +8,20 @@
 # pylint: disable=unused-variable
 # pylint: disable=too-many-arguments
 
+from typing import Annotated
 
 from _common import assert_handler_signature_against_model
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends, status
 from models_library.api_schemas_webserver.wallets import WalletGet
 from models_library.generics import Envelope
 from models_library.projects import ProjectID
 from models_library.wallets import WalletID
 from simcore_service_webserver._meta import API_VTAG
 from simcore_service_webserver.projects._common.models import ProjectPathParams
+from simcore_service_webserver.projects._wallets_handlers import (
+    _PayProjectDebtBody,
+    _ProjectWalletPathParams,
+)
 
 router = APIRouter(
     prefix=f"/{API_VTAG}",
@@ -51,3 +56,17 @@ async def connect_wallet_to_project(
 
 
 assert_handler_signature_against_model(connect_wallet_to_project, ProjectPathParams)
+
+
+@router.post(
+    "/projects/{project_id}/wallet/{wallet_id}:pay-debt",
+    status_code=status.HTTP_204_NO_CONTENT,
+)
+async def pay_project_debt(
+    _path: Annotated[_ProjectWalletPathParams, Depends()],
+    _body: Annotated[_PayProjectDebtBody, Depends()],
+):
+    ...
+
+
+assert_handler_signature_against_model(connect_wallet_to_project, ProjectPathParams)

packages/models-library/src/models_library/resource_tracker.py

@@ -38,16 +38,50 @@ class ServiceRunStatus(StrAutoEnum):
 
 
 class CreditTransactionStatus(StrAutoEnum):
+    # Represents the possible statuses of a credit transaction.
+
     PENDING = auto()
+    # The transaction is pending and has not yet been finalized.
+    # Example: During the running of a service, the transaction remains in the Pending state until the service is stopped.
+
     BILLED = auto()
+    # The transaction has been successfully billed.
+
+    IN_DEBT = auto()
+    # The transaction is marked as in debt.
+    # Example: This occurs when a computational job continues to run even though the user does not have sufficient credits in their wallet.
+
     NOT_BILLED = auto()
+    # The transaction will not be billed.
+    # Example: This status is used when there is an issue on our side, and we decide not to bill the user.
+
     REQUIRES_MANUAL_REVIEW = auto()
+    # The transaction requires manual review due to potential issues.
+    # NOTE: This status is currently not in use.
 
 
 class CreditClassification(StrAutoEnum):
-    ADD_WALLET_TOP_UP = auto()  # user top up credits
-    DEDUCT_SERVICE_RUN = auto()  # computational/dynamic service run costs)
+    # Represents the different types of credit classifications.
+
+    ADD_WALLET_TOP_UP = auto()
+    # Indicates that credits have been added to the user's wallet through a top-up.
+    # Example: The user adds funds to their wallet to increase their available credits.
+
+    DEDUCT_SERVICE_RUN = auto()
+    # Represents a deduction from the user's wallet due to the costs of running a computational or dynamic service.
+    # Example: Credits are deducted when the user runs a simulation.
+
     DEDUCT_LICENSE_PURCHASE = auto()
+    # Represents a deduction from the user's wallet for purchasing a license.
+    # Example: The user purchases a license to access premium features such as VIP models.
+
+    ADD_WALLET_EXCHANGE = auto()
+    # Represents the addition of credits to the user's wallet through an exchange.
+    # Example: Credits are added due to credit exchange between wallets.
+
+    DEDUCT_WALLET_EXCHANGE = auto()
+    # Represents a deduction of credits from the user's wallet through an exchange.
+    # Example: Credits are deducted due to credit exchange between wallets.
 
 
 class PricingPlanClassification(StrAutoEnum):

packages/postgres-database/src/simcore_postgres_database/migration/versions/a3a58471b0f1_add_credit_transaction_classification_.py

@@ -0,0 +1,35 @@
+"""add credit transaction classification enums
+
+Revision ID: a3a58471b0f1
+Revises: f19905923355
+Create Date: 2025-01-14 13:44:05.025647+00:00
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "a3a58471b0f1"
+down_revision = "f19905923355"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.execute(sa.DDL("ALTER TYPE credittransactionstatus ADD VALUE 'IN_DEBT'"))
+    op.execute(
+        sa.DDL(
+            "ALTER TYPE credittransactionclassification ADD VALUE 'ADD_WALLET_EXCHANGE'"
+        )
+    )
+    op.execute(
+        sa.DDL(
+            "ALTER TYPE credittransactionclassification ADD VALUE 'DEDUCT_WALLET_EXCHANGE'"
+        )
+    )
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###

packages/postgres-database/src/simcore_postgres_database/models/resource_tracker_credit_transactions.py

@@ -18,6 +18,7 @@
 class CreditTransactionStatus(str, enum.Enum):
     PENDING = "PENDING"
     BILLED = "BILLED"
+    IN_DEBT = "IN_DEBT"
     NOT_BILLED = "NOT_BILLED"
     REQUIRES_MANUAL_REVIEW = "REQUIRES_MANUAL_REVIEW"
 
@@ -28,6 +29,8 @@ class CreditTransactionClassification(str, enum.Enum):
         "DEDUCT_SERVICE_RUN"  # computational/dynamic service run costs)
     )
     DEDUCT_LICENSE_PURCHASE = "DEDUCT_LICENSE_PURCHASE"
+    ADD_WALLET_EXCHANGE = "ADD_WALLET_EXCHANGE"
+    DEDUCT_WALLET_EXCHANGE = "DEDUCT_WALLET_EXCHANGE"
 
 
 resource_tracker_credit_transactions = sa.Table(

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/resource_usage_tracker/credit_transactions.py

@@ -0,0 +1,84 @@
+import logging
+from typing import Final
+
+from models_library.api_schemas_resource_usage_tracker import (
+    RESOURCE_USAGE_TRACKER_RPC_NAMESPACE,
+)
+from models_library.api_schemas_resource_usage_tracker.credit_transactions import (
+    CreditTransactionCreateBody,
+    WalletTotalCredits,
+)
+from models_library.products import ProductName
+from models_library.projects import ProjectID
+from models_library.rabbitmq_basic_types import RPCMethodName
+from models_library.resource_tracker import CreditTransactionStatus
+from models_library.wallets import WalletID
+from pydantic import NonNegativeInt, TypeAdapter
+
+from ....logging_utils import log_decorator
+from ....rabbitmq import RabbitMQRPCClient
+
+_logger = logging.getLogger(__name__)
+
+
+_DEFAULT_TIMEOUT_S: Final[NonNegativeInt] = 20
+
+_RPC_METHOD_NAME_ADAPTER: TypeAdapter[RPCMethodName] = TypeAdapter(RPCMethodName)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def get_wallet_total_credits(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    product_name: ProductName,
+    wallet_id: WalletID,
+) -> WalletTotalCredits:
+    result = await rabbitmq_rpc_client.request(
+        RESOURCE_USAGE_TRACKER_RPC_NAMESPACE,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("get_wallet_total_credits"),
+        product_name=product_name,
+        wallet_id=wallet_id,
+        timeout_s=_DEFAULT_TIMEOUT_S,
+    )
+    assert isinstance(result, WalletTotalCredits)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def get_project_wallet_total_credits(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    product_name: ProductName,
+    wallet_id: WalletID,
+    project_id: ProjectID,
+    transaction_status: CreditTransactionStatus | None = None,
+) -> WalletTotalCredits:
+    result = await rabbitmq_rpc_client.request(
+        RESOURCE_USAGE_TRACKER_RPC_NAMESPACE,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("get_project_wallet_total_credits"),
+        product_name=product_name,
+        wallet_id=wallet_id,
+        project_id=project_id,
+        transaction_status=transaction_status,
+        timeout_s=_DEFAULT_TIMEOUT_S,
+    )
+    assert isinstance(result, WalletTotalCredits)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def pay_project_debt(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    project_id: ProjectID,
+    current_wallet_transaction: CreditTransactionCreateBody,
+    new_wallet_transaction: CreditTransactionCreateBody,
+) -> None:
+    await rabbitmq_rpc_client.request(
+        RESOURCE_USAGE_TRACKER_RPC_NAMESPACE,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("pay_project_debt"),
+        project_id=project_id,
+        current_wallet_transaction=current_wallet_transaction,
+        new_wallet_transaction=new_wallet_transaction,
+        timeout_s=_DEFAULT_TIMEOUT_S,
+    )

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/resource_usage_tracker/errors.py

@@ -27,3 +27,10 @@ class LicensedItemCheckoutNotFoundError(LicensesBaseError):
     CanNotCheckoutServiceIsNotRunningError,
     LicensedItemCheckoutNotFoundError,
 )
+
+
+### Transaction Error
+
+
+class WalletTransactionError(OsparcErrorMixin, Exception):
+    msg_template = "{msg}"

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/resource_usage_tracker/service_runs.py

@@ -1,3 +1,4 @@
+# pylint: disable=too-many-arguments
 import logging
 from typing import Final
 
@@ -9,8 +10,10 @@
     ServiceRunPage,
 )
 from models_library.products import ProductName
+from models_library.projects import ProjectID
 from models_library.rabbitmq_basic_types import RPCMethodName
 from models_library.resource_tracker import (
+    CreditTransactionStatus,
     ServiceResourceUsagesFilters,
     ServicesAggregatedUsagesTimePeriod,
     ServicesAggregatedUsagesType,
@@ -37,24 +40,30 @@ async def get_service_run_page(
     *,
     user_id: UserID,
     product_name: ProductName,
-    limit: int = 20,
-    offset: int = 0,
     wallet_id: WalletID | None = None,
     access_all_wallet_usage: bool = False,
-    order_by: OrderBy | None = None,
     filters: ServiceResourceUsagesFilters | None = None,
+    transaction_status: CreditTransactionStatus | None = None,
+    project_id: ProjectID | None = None,
+    # pagination
+    offset: int = 0,
+    limit: int = 20,
+    # ordering
+    order_by: OrderBy | None = None,
 ) -> ServiceRunPage:
     result = await rabbitmq_rpc_client.request(
         RESOURCE_USAGE_TRACKER_RPC_NAMESPACE,
         _RPC_METHOD_NAME_ADAPTER.validate_python("get_service_run_page"),
         user_id=user_id,
         product_name=product_name,
-        limit=limit,
-        offset=offset,
         wallet_id=wallet_id,
         access_all_wallet_usage=access_all_wallet_usage,
-        order_by=order_by,
         filters=filters,
+        transaction_status=transaction_status,
+        project_id=project_id,
+        offset=offset,
+        limit=limit,
+        order_by=order_by,
         timeout_s=_DEFAULT_TIMEOUT_S,
     )
     assert isinstance(result, ServiceRunPage)  # nosec

services/efs-guardian/src/simcore_service_efs_guardian/services/background_tasks.py

@@ -45,11 +45,12 @@ async def removal_policy_task(app: FastAPI) -> None:
             _project_last_change_date = (
                 await projects_repo.get_project_last_change_date(project_id)
             )
-        except DBProjectNotFoundError as exc:
-            _logger.warning(
-                "Project %s not found, this should not happen, please investigate (contact MD)",
-                exc.msg_template,
+        except DBProjectNotFoundError:
+            _logger.info(
+                "Project %s not found. Removing EFS data for project {project_id} started",
+                project_id,
             )
+            await efs_manager.remove_project_efs_data(project_id)
         if (
             _project_last_change_date
             < base_start_timestamp

services/efs-guardian/src/simcore_service_efs_guardian/services/fire_and_forget_setup.py

@@ -2,6 +2,7 @@
 from collections.abc import Awaitable, Callable
 
 from fastapi import FastAPI
+from servicelib.async_utils import cancel_wait_task
 from servicelib.logging_utils import log_catch, log_context
 
 _logger = logging.getLogger(__name__)
@@ -27,7 +28,7 @@ async def _stop() -> None:
             assert _app  # nosec
             if _app.state.efs_guardian_fire_and_forget_tasks:
                 for task in _app.state.efs_guardian_fire_and_forget_tasks:
-                    task.cancel()
+                    await cancel_wait_task(task)
 
     return _stop
 

services/resource-usage-tracker/src/simcore_service_resource_usage_tracker/api/rest/_resource_tracker.py

@@ -34,7 +34,7 @@
 async def get_credit_transactions_sum(
     wallet_total_credits: Annotated[
         WalletTotalCredits,
-        Depends(credit_transactions.sum_credit_transactions_by_product_and_wallet),
+        Depends(credit_transactions.sum_wallet_credits),
     ],
 ):
     return wallet_total_credits