primary email change functionality

backend/app/services/email_addresses.py

@@ -0,0 +1,73 @@
+## Copyright © 2026 Olaf Kolkman
+## SPDX-License-Identifier: GPL-3.0-or-later
+
+from datetime import datetime, timedelta, timezone
+from hashlib import sha256
+from secrets import token_urlsafe
+from uuid import uuid4
+
+from backend.app.core.config import settings
+from backend.app.database import get_connection
+from backend.app.services.email_service import send_verification_email
+
+
+def list_user_email_addresses(user_id: str) -> list[dict]:
+    with get_connection() as conn:
+        rows = conn.execute(
+            'SELECT id, email, verified, created_at FROM user_email_addresses WHERE user_id = ? ORDER BY created_at',
+            (user_id,),
+        ).fetchall()
+    return [dict(row) | {'verified': bool(row['verified']), 'can_be_primary': bool(row['verified'])} for row in rows]
+
+
+def add_user_email_address(user_id: str, email: str) -> dict:
+    email = email.strip().lower()
+    if not email:
+        raise ValueError('Email address is required')
+    with get_connection() as conn:
+        try:
+            row = conn.execute(
+                'INSERT INTO user_email_addresses (id, user_id, email) VALUES (?, ?, ?) RETURNING id, email, verified, created_at',
+                (str(uuid4()), user_id, email),
+            ).fetchone()
+            conn.commit()
+        except Exception as error:
+            if 'UNIQUE constraint failed' in str(error):
+                raise ValueError('Email address already exists') from error
+            raise
+    return dict(row) | {'verified': bool(row['verified']), 'can_be_primary': bool(row['verified'])}
+
+
+def create_email_verification(user_id: str, address_id: str) -> tuple[str, str]:
+    token = token_urlsafe(32)
+    expires_at = datetime.now(timezone.utc) + timedelta(hours=settings.email_verification_expiry_hours)
+    with get_connection() as conn:
+        address = conn.execute(
+            'SELECT email FROM user_email_addresses WHERE id = ? AND user_id = ?',
+            (address_id, user_id),
+        ).fetchone()
+        if address is None:
+            raise ValueError('Email address not found')
+        conn.execute('DELETE FROM email_address_verification_tokens WHERE email_address_id = ?', (address_id,))
+        conn.execute(
+            'INSERT INTO email_address_verification_tokens (id, email_address_id, token_hash, expires_at) VALUES (?, ?, ?, ?)',
+            (str(uuid4()), address_id, sha256(token.encode()).hexdigest(), expires_at.isoformat()),
+        )
+        conn.commit()
+    return address['email'], token
+
+
+def verify_user_email_address(token: str) -> bool:
+    now = datetime.now(timezone.utc).isoformat()
+    with get_connection() as conn:
+        row = conn.execute(
+            '''SELECT email_address_id FROM email_address_verification_tokens
+               WHERE token_hash = ? AND expires_at > ?''',
+            (sha256(token.encode()).hexdigest(), now),
+        ).fetchone()
+        if row is None:
+            return False
+        conn.execute('UPDATE user_email_addresses SET verified = 1, updated_at = CURRENT_TIMESTAMP WHERE id = ?', (row['email_address_id'],))
+        conn.execute('DELETE FROM email_address_verification_tokens WHERE email_address_id = ?', (row['email_address_id'],))
+        conn.commit()
+    return True