wiggleverse-ecomm/backend/app/domains/accounts/service.py

"""accounts — the identity service (SD-0001 §6.5).

All identity rules live here exactly once (INV-6): email normalization (INV-2); one-time
code issue/verify handled like a secret (INV-3 — hashed with the app secret as pepper,
10-minute TTL, single-use, ≤5 attempts, 60s resend cooldown); and get-or-create account.
Uniform for new vs known emails so the surface never enumerates accounts (§6.6). Imports
only app.platform (downward); takes the connection and mailer as parameters (the BFF wires
them) so the domain owns no request/transport concerns.
"""
from __future__ import annotations

import hashlib
import hmac
import re
import secrets
from datetime import datetime, timedelta, timezone

import psycopg

from app.platform import config
from app.platform.mailer import Mailer

from .errors import CodeExhausted, CodeExpired, CodeMismatch, InvalidEmail, ResendCooldown
from .models import Account

# INV-3 constants — the one-time-code policy.
CODE_TTL = timedelta(minutes=10)
RESEND_COOLDOWN = timedelta(seconds=60)
MAX_ATTEMPTS = 5
_CODE_DIGITS = 6

# A pragmatic email check — rejects obvious non-addresses without an email-validator
# dependency. The provable test of an address is that the emailed code comes back (verify).
_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")


def normalize_email(email: str) -> str:
    """Lowercase + strip — the canonical form one account is keyed by (INV-2)."""
    norm = email.strip().lower()
    if not _EMAIL_RE.match(norm):
        raise InvalidEmail(email)
    return norm


def _hash_code(code: str) -> str:
    """HMAC-SHA256 of the code, peppered by the app secret (INV-3 — hashes only, not the code)."""
    key = config.session_secret().encode("utf-8")
    return hmac.new(key, code.encode("utf-8"), hashlib.sha256).hexdigest()


def _now() -> datetime:
    return datetime.now(timezone.utc)


def request_code(conn: psycopg.Connection, mailer: Mailer, email: str) -> None:
    """Issue a one-time code to `email` and dispatch it. Uniform for new/known emails (§6.6).

    Enforces the 60s resend cooldown (INV-3 → PUC-2c). Opportunistically purges this email's
    spent/expired codes. Never reveals whether the email already has an account.
    """
    email = normalize_email(email)
    now = _now()

    last = conn.execute(
        "SELECT max(created_at) FROM auth_code WHERE email = %s AND consumed_at IS NULL",
        (email,),
    ).fetchone()[0]
    if last is not None:
        elapsed = now - last
        if elapsed < RESEND_COOLDOWN:
            retry_after_s = int((RESEND_COOLDOWN - elapsed).total_seconds()) + 1
            raise ResendCooldown(retry_after_s)

    # Housekeeping: drop this email's consumed/expired codes (§6.3) before issuing a fresh one.
    conn.execute(
        "DELETE FROM auth_code WHERE email = %s AND (consumed_at IS NOT NULL OR expires_at < %s)",
        (email, now),
    )

    code = "".join(secrets.choice("0123456789") for _ in range(_CODE_DIGITS))
    conn.execute(
        "INSERT INTO auth_code (email, code_hash, expires_at) VALUES (%s, %s, %s)",
        (email, _hash_code(code), now + CODE_TTL),
    )
    conn.commit()

    mailer.send(
        to=email,
        subject=f"Your ecomm code: {code}",
        body=(
            f"Your ecomm one-time code is {code}.\n"
            f"It is valid for {int(CODE_TTL.total_seconds() // 60)} minutes.\n"
            "If you didn't request this, ignore this message."
        ),
    )


def verify(conn: psycopg.Connection, email: str, code: str) -> tuple[Account, bool]:
    """Verify a code and resolve the account. Returns (account, created).

    Consumes the live code transactionally (single-use, INV-3): wrong codes spend an attempt
    (PUC-2a) until MAX_ATTEMPTS invalidates it (CodeExhausted); expired/absent codes raise
    CodeExpired (PUC-2b). On success the code is consumed and the account is got-or-created
    by normalized email (INV-2). Account creation IS email verification (the address
    provably received the code).
    """
    email = normalize_email(email)
    now = _now()

    row = conn.execute(
        "SELECT id, code_hash, expires_at, attempts FROM auth_code "
        "WHERE email = %s AND consumed_at IS NULL "
        "ORDER BY created_at DESC LIMIT 1 FOR UPDATE",
        (email,),
    ).fetchone()
    if row is None:
        conn.commit()
        raise CodeExpired(email)  # nothing live to verify — offer a fresh code
    code_id, code_hash, expires_at, attempts = row

    if expires_at < now:
        conn.commit()
        raise CodeExpired(email)

    if not hmac.compare_digest(code_hash, _hash_code(code)):
        attempts += 1
        if attempts >= MAX_ATTEMPTS:
            conn.execute(
                "UPDATE auth_code SET attempts = %s, consumed_at = %s WHERE id = %s",
                (attempts, now, code_id),
            )
            conn.commit()
            raise CodeExhausted(email)
        conn.execute("UPDATE auth_code SET attempts = %s WHERE id = %s", (attempts, code_id))
        conn.commit()
        raise CodeMismatch(attempts_remaining=MAX_ATTEMPTS - attempts)

    # Correct: consume the code and get-or-create the account in the same transaction.
    conn.execute("UPDATE auth_code SET consumed_at = %s WHERE id = %s", (now, code_id))
    account, created = _get_or_create_account(conn, email)
    conn.commit()
    return account, created


def _get_or_create_account(conn: psycopg.Connection, email: str) -> tuple[Account, bool]:
    """Insert the account if absent; return (account, created). Race-safe via the unique index."""
    inserted = conn.execute(
        "INSERT INTO account (email) VALUES (%s) ON CONFLICT (email) DO NOTHING RETURNING id",
        (email,),
    ).fetchone()
    if inserted is not None:
        return Account(id=inserted[0], email=email), True
    existing = conn.execute("SELECT id FROM account WHERE email = %s", (email,)).fetchone()
    return Account(id=existing[0], email=email), False


def get_account(conn: psycopg.Connection, account_id: int) -> Account | None:
    """Load an account by id (for the signed-in `/me`)."""
    row = conn.execute("SELECT id, email FROM account WHERE id = %s", (account_id,)).fetchone()
    return Account(id=row[0], email=row[1]) if row else None