✨ Implement ConfirmationRepository for managing user confirmation tokens

Author: pcrespov

services/web/server/src/simcore_service_webserver/login/_confirmation_repository.py

@@ -0,0 +1,147 @@
+import logging
+from datetime import datetime
+from typing import Any
+
+import sqlalchemy as sa
+from models_library.users import UserID
+from servicelib.utils_secrets import generate_passcode
+from simcore_postgres_database.models.confirmations import confirmations
+from simcore_postgres_database.models.users import users
+from simcore_postgres_database.utils_repos import (
+    pass_or_acquire_connection,
+    transaction_context,
+)
+from sqlalchemy.engine import Row
+from sqlalchemy.ext.asyncio import AsyncConnection
+
+from ..db.base_repository import BaseRepository
+from ._models import ActionLiteralStr, Confirmation
+
+_logger = logging.getLogger(__name__)
+
+
+def _to_domain(confirmation_row: Row) -> Confirmation:
+    return Confirmation.model_validate(confirmation_row)
+
+
+class ConfirmationRepository(BaseRepository):
+
+    async def create_confirmation(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        user_id: UserID,
+        action: ActionLiteralStr,
+        data: str | None = None,
+    ) -> Confirmation:
+        """Create a new confirmation token for a user action."""
+        async with pass_or_acquire_connection(self.engine, connection) as conn:
+            # Generate unique code
+            while True:
+                # NOTE: use only numbers since front-end does not handle well url encoding
+                numeric_code: str = generate_passcode(20)
+
+                # Check if code already exists
+                check_query = sa.select(confirmations.c.code).where(
+                    confirmations.c.code == numeric_code
+                )
+                result = await conn.execute(check_query)
+                if result.one_or_none() is None:
+                    break
+
+            # Insert confirmation
+            insert_query = (
+                sa.insert(confirmations)
+                .values(
+                    code=numeric_code,
+                    user_id=user_id,
+                    action=action,
+                    data=data,
+                    created_at=datetime.utcnow(),
+                )
+                .returning(*confirmations.c)
+            )
+
+            result = await conn.execute(insert_query)
+            row = result.one()
+            return _to_domain(row)
+
+    async def get_confirmation(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        filter_dict: dict[str, Any],
+    ) -> Confirmation | None:
+        """Get a confirmation by filter criteria."""
+        # Handle legacy "user" key
+        if "user" in filter_dict:
+            filter_dict["user_id"] = filter_dict.pop("user")["id"]
+
+        # Build where conditions
+        where_conditions = []
+        for key, value in filter_dict.items():
+            if hasattr(confirmations.c, key):
+                where_conditions.append(getattr(confirmations.c, key) == value)
+
+        query = sa.select(*confirmations.c).where(sa.and_(*where_conditions))
+
+        async with pass_or_acquire_connection(self.engine, connection) as conn:
+            result = await conn.execute(query)
+            if row := result.one_or_none():
+                return _to_domain(row)
+            return None
+
+    async def delete_confirmation(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        confirmation: Confirmation,
+    ) -> None:
+        """Delete a confirmation token."""
+        query = sa.delete(confirmations).where(
+            confirmations.c.code == confirmation.code
+        )
+
+        async with pass_or_acquire_connection(self.engine, connection) as conn:
+            await conn.execute(query)
+
+    async def delete_confirmation_and_user(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        user_id: UserID,
+        confirmation: Confirmation,
+    ) -> None:
+        """Atomically delete confirmation and user."""
+        async with transaction_context(self.engine, connection) as conn:
+            # Delete confirmation
+            await conn.execute(
+                sa.delete(confirmations).where(
+                    confirmations.c.code == confirmation.code
+                )
+            )
+
+            # Delete user
+            await conn.execute(sa.delete(users).where(users.c.id == user_id))
+
+    async def delete_confirmation_and_update_user(
+        self,
+        connection: AsyncConnection | None = None,
+        *,
+        user_id: UserID,
+        updates: dict[str, Any],
+        confirmation: Confirmation,
+    ) -> None:
+        """Atomically delete confirmation and update user."""
+        async with transaction_context(self.engine, connection) as conn:
+            # Delete confirmation
+            await conn.execute(
+                sa.delete(confirmations).where(
+                    confirmations.c.code == confirmation.code
+                )
+            )
+
+            # Update user
+            await conn.execute(
+                sa.update(users).where(users.c.id == user_id).values(**updates)
+            )

services/web/server/src/simcore_service_webserver/login/_login_repository_legacy.py

@@ -125,6 +125,9 @@ async def delete_confirmation_and_update_user(
                 conn, self.user_tbl, {"id": user_id}, updates
             )
 
+    # NOTE: This class is deprecated. Use ConfirmationRepository instead.
+    # Keeping for backwards compatibility during migration.
+
 
 def get_plugin_storage(app: web.Application) -> AsyncpgStorage:
     storage = cast(AsyncpgStorage, app.get(APP_LOGIN_STORAGE_KEY))

services/web/server/src/simcore_service_webserver/login/_models.py

@@ -1,9 +1,26 @@
 from collections.abc import Callable
+from datetime import datetime
+from typing import Literal
 
+from models_library.users import UserID
 from pydantic import BaseModel, ConfigDict, SecretStr, ValidationInfo
 
 from .constants import MSG_PASSWORD_MISMATCH
 
+ActionLiteralStr = Literal[
+    "REGISTRATION", "INVITATION", "RESET_PASSWORD", "CHANGE_EMAIL"
+]
+
+
+class Confirmation(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    code: str
+    user_id: UserID
+    action: ActionLiteralStr
+    data: str | None
+    created_at: datetime
+
 
 class InputSchema(BaseModel):
     model_config = ConfigDict(