diff --git a/backend/app/gateway/auth/workspace_slug.py b/backend/app/gateway/auth/workspace_slug.py new file mode 100644 index 00000000..da399cd1 --- /dev/null +++ b/backend/app/gateway/auth/workspace_slug.py @@ -0,0 +1,83 @@ +"""Workspace slug helpers for the registration / initialize flow. + +Stage 0 PR4 T4.10. Two responsibilities: + +1. ``auto_slug_from_email(email)`` — pure transform from an email's + local part to a base slug that matches the schema's + ``^[a-z0-9](-?[a-z0-9])*$`` pattern. +2. ``next_available_slug(base, exists_check=...)`` — collision walker + that appends ``-2``, ``-3``, … until ``exists_check`` reports the + candidate is free. Kept separate from ``auto_slug_from_email`` so + the pure function can be tested without a database. + +Lives in the auth package (not in ``persistence``) because the input +is the user's email — a registration-time concept that doesn't belong +in a generic ``WorkspaceRepository``. +""" + +from __future__ import annotations + +import re +import secrets +from collections.abc import Awaitable, Callable + +# Mirror the schema's slug rules from +# ``deerflow.persistence.workspace.sql`` so callers of this module +# never need to import private constants from persistence. +_SLUG_MIN_LEN = 3 +_SLUG_MAX_LEN = 32 + + +def auto_slug_from_email(email: str) -> str: + """Map an email to a deterministic, schema-valid base slug. + + Algorithm (from workspace-schema-design §3.1): + + 1. Take the local part (before ``@``). + 2. Replace ``+``, ``_``, ``.`` with ``-`` and lowercase. + 3. Strip everything that isn't ``[a-z0-9-]``. + 4. Collapse repeated ``-``; strip leading/trailing ``-``. + 5. Clamp to 32 chars. + 6. If the result is shorter than the schema minimum (3 chars) or + empty, fall back to ``user-{token_hex(4)}`` so we always emit + a valid slug. + + The returned slug is the *base* — callers must run it through + :func:`next_available_slug` before persisting to handle collisions. + """ + local = email.split("@", 1)[0] + local = re.sub(r"[+_.]", "-", local).lower() + local = re.sub(r"[^a-z0-9-]", "", local) + local = re.sub(r"-+", "-", local).strip("-") + slug = local[:_SLUG_MAX_LEN] + if len(slug) < _SLUG_MIN_LEN: + return f"user-{secrets.token_hex(4)}" + return slug + + +async def next_available_slug( + base: str, + *, + exists_check: Callable[[str], Awaitable[bool]], +) -> str: + """Return the first of ``base``, ``base-2``, ``base-3``, … that ``exists_check`` reports free. + + Caller-supplied ``exists_check`` is awaited once per candidate so + we can swap in a repository's ``get_by_slug`` without coupling + this module to persistence imports. + + When ``base + '-N'`` would exceed the 32-char schema limit, the + base is truncated before the suffix is appended. The walker never + returns an over-long slug. + """ + if not await exists_check(base): + return base + + n = 2 + while True: + suffix = f"-{n}" + max_base_len = _SLUG_MAX_LEN - len(suffix) + candidate = f"{base[:max_base_len]}{suffix}" + if not await exists_check(candidate): + return candidate + n += 1 diff --git a/backend/tests/test_workspace_slug.py b/backend/tests/test_workspace_slug.py new file mode 100644 index 00000000..fac231cb --- /dev/null +++ b/backend/tests/test_workspace_slug.py @@ -0,0 +1,98 @@ +"""Slug helpers for registration / initialize (Stage 0 PR4 T4.10). + +Two surfaces under test: + +- ``auto_slug_from_email`` — pure transform; no DB dependency. +- ``next_available_slug`` — async collision walker; we stub the + ``exists_check`` callable so the test stays a unit test. +""" + +from __future__ import annotations + +import re + +import pytest + +from app.gateway.auth.workspace_slug import auto_slug_from_email, next_available_slug + +# Mirror of the schema's slug pattern. Keeping it inline keeps this +# test self-contained — if the schema regex changes we want this test +# to refuse to lie about validity. +_SLUG_PATTERN = re.compile(r"^[a-z0-9](-?[a-z0-9])*$") + + +@pytest.mark.parametrize( + ("email", "expected"), + [ + ("foo@example.com", "foo"), + ("foo.bar@example.com", "foo-bar"), + ("foo+spam@example.com", "foo-spam"), + ("foo_bar@example.com", "foo-bar"), + ("Foo.Bar@example.com", "foo-bar"), + ("foo.bar+spam@example.com", "foo-bar-spam"), + ("aaaaaaaaaabbbbbbbbbbccccccccccddddd@example.com", "aaaaaaaaaabbbbbbbbbbccccccccccdd"), + ], +) +def test_auto_slug_known_inputs(email: str, expected: str) -> None: + """Deterministic mapping for the inputs called out in the design doc.""" + assert auto_slug_from_email(email) == expected + assert _SLUG_PATTERN.fullmatch(auto_slug_from_email(email)), "schema regex must accept the output" + + +@pytest.mark.parametrize( + "email", + [ + "@example.com", # no local part + "a@example.com", # too short + "ab@example.com", # still too short + "...+_+...@example.com", # only separators + "---@example.com", # only hyphens + "🎉@example.com", # non-ASCII + ], +) +def test_auto_slug_falls_back_when_unusable(email: str) -> None: + """Pathological emails fall back to ``user-{token}`` so the slug is always valid.""" + slug = auto_slug_from_email(email) + assert slug.startswith("user-"), f"expected fallback, got {slug!r}" + assert _SLUG_PATTERN.fullmatch(slug) + assert 3 <= len(slug) <= 32 + + +@pytest.mark.anyio +@pytest.mark.parametrize("anyio_backend", ["asyncio"]) +async def test_next_available_slug_returns_base_when_free(anyio_backend) -> None: + """No collision → ``base`` is returned unchanged.""" + seen: set[str] = set() + + async def exists(s: str) -> bool: + return s in seen + + assert await next_available_slug("foo", exists_check=exists) == "foo" + + +@pytest.mark.anyio +@pytest.mark.parametrize("anyio_backend", ["asyncio"]) +async def test_next_available_slug_walks_through_collisions(anyio_backend) -> None: + """``foo``, ``foo-2`` taken → walker lands on ``foo-3``.""" + seen = {"foo", "foo-2"} + + async def exists(s: str) -> bool: + return s in seen + + assert await next_available_slug("foo", exists_check=exists) == "foo-3" + + +@pytest.mark.anyio +@pytest.mark.parametrize("anyio_backend", ["asyncio"]) +async def test_next_available_slug_truncates_base_to_fit_suffix(anyio_backend) -> None: + """A 32-char base + ``-2`` would exceed the limit → base is shortened.""" + base = "a" * 32 # exactly at the limit + seen = {base} + + async def exists(s: str) -> bool: + return s in seen + + result = await next_available_slug(base, exists_check=exists) + assert len(result) <= 32 + assert result.endswith("-2") + assert _SLUG_PATTERN.fullmatch(result)