feat(custom-domains): public API + dashboard + tenant routing policy

Brings custom domains from a service-layer-only feature to user-facing.
Owners can now register a domain (e.g. links.acme.com), publish DNS
records returned by the API, verify, and serve short links under their
own brand.

API
  POST   /api/v1/custom-domains              register
  POST   /api/v1/custom-domains/{id}/verify  trigger verification
  GET    /api/v1/custom-domains              list owned (paginated)
  DELETE /api/v1/custom-domains/{id}         revoke (?cascade=true also
                                             bulk-deletes URLs on the domain)
  DELETE /api/v1/urls?domain=<fqdn>          bulk URL delete on owned domain

POST /api/v1/shorten accepts an optional `domain` field; ownership +
ACTIVE-status enforced at the route layer. GET /api/v1/urls accepts a
`?domain=` filter and UrlListItem now exposes `domain`.

Tenant routing
Custom hostnames serve a strict allowlist: `/<alias>`, `/<alias>/password`,
`/favicon.ico`, `/robots.txt`. Everything else (operator surface like
/api, /dashboard, /auth, /oauth; brand pages like /about, /contact;
the stats suffix `/<alias>+`; /report; /health) returns 404 so
customer-branded hosts don't expose the canonical app surface.

Per-tenant `/robots.txt` is served inline as `Disallow: /`, and every
custom-tenant response is stamped with
`X-Robots-Tag: noindex, nofollow, noarchive`. The existing system-default
`/<alias>` redirect header gained `noarchive` alongside this.

Auth + scopes
New API key scopes `domains:manage` and `domains:read`. New
`require_scopes_verified` dependency combines scope auth with the
email-verified gate so both JWT and API key callers can create.

Response shape
CustomDomainResponse now carries structured `dns_records` (each with
type/name/value/purpose) plus `setup_notes` for human-readable warnings
(e.g. "Cloudflare DNS detected — set DNS-only / grey cloud"). Dropped
the old free-form `setup_instructions` string and the internal
`verification_token` field (no longer surfaced over the wire).

Service / repository
CustomDomainService gains `assert_owned`, `assert_owned_and_active`
helpers and `delete(*, cascade=False)` returning `(doc, urls_deleted)`.
Cascade order: REVOKE first, bulk delete URLs second, edge eviction
third, cache invalidate last. Partial failures swallow + log; a future
garbage-collection worker reaps orphan URLs.

UrlService.create accepts an opaque `domain=`; alias-uniqueness now
scopes through `(domain, alias)`. `delete_all_by_domain` refuses the
system-default namespace defensively.

Repository: list_aliases_by_owner_and_domain + delete_many_by_owner_and_domain.
Cache: invalidate_many for bulk cleanup.

Rate limits
Slowapi limits on every new route (5/hr CREATE, 10/min VERIFY/DELETE,
60/min READ, 5/min bulk delete). Service-side per-user Mongo + Redis
quotas still apply on top — layered defense.

Dashboard
New /dashboard/domains page with a sidebar nav entry. Markup matches
the existing keys/links pattern: external CSS+JS, `<template>` row
cloning, status badges via design tokens, server shell + client fetch
hydration. Delete modal fetches the current URL count and offers the
caller a choice between orphaning the URLs or cascade-deleting them.

Tests
74 new test cases across DTOs, service helpers, route happy/sad paths,
URL repository bulk methods, cache invalidate_many, and the tenant
middleware routing matrix (including robots.txt, favicon, operator
blocklist, stats-suffix block). Full suite: 1619 passing.

Author: Zingzy

dependencies/__init__.py

@@ -6,6 +6,8 @@
 """
 
 from dependencies.auth import (
+    DOMAIN_MANAGE_SCOPES,
+    DOMAIN_READ_SCOPES,
     SHORTEN_SCOPES,
     STATS_SCOPES,
     URL_MANAGEMENT_SCOPES,
@@ -24,6 +26,7 @@
     require_jwt,
     require_jwt_verified,
     require_scopes,
+    require_scopes_verified,
     require_verified_email,
 )
 from dependencies.infra import (
@@ -46,8 +49,10 @@
     ClickSvc,
     ContactSvc,
     CredentialSvc,
+    CustomDomainSvc,
     DeviceAuthSvc,
     ExportSvc,
+    FeatureFlagSvc,
     OAuthSvc,
     PasswordSvc,
     ProfilePictureSvc,
@@ -61,8 +66,10 @@
     get_click_service,
     get_contact_service,
     get_credential_service,
+    get_custom_domain_service,
     get_device_auth_service,
     get_export_service,
+    get_feature_flag_service,
     get_oauth_service,
     get_password_service,
     get_profile_picture_service,
@@ -73,6 +80,8 @@
 )
 
 __all__ = [
+    "DOMAIN_MANAGE_SCOPES",
+    "DOMAIN_READ_SCOPES",
     "SHORTEN_SCOPES",
     "STATS_SCOPES",
     "URL_MANAGEMENT_SCOPES",
@@ -88,8 +97,10 @@
     "ContactSvc",
     "CredentialSvc",
     "CurrentUser",
+    "CustomDomainSvc",
     "DeviceAuthSvc",
     "ExportSvc",
+    "FeatureFlagSvc",
     "JwtConfig",
     "JwtUser",
     "JwtVerifiedUser",
@@ -114,10 +125,12 @@
     "get_contact_service",
     "get_credential_service",
     "get_current_user",
+    "get_custom_domain_service",
     "get_db",
     "get_device_auth_service",
     "get_email_provider",
     "get_export_service",
+    "get_feature_flag_service",
     "get_geoip_service",
     "get_jwt_config",
     "get_oauth_providers",
@@ -136,5 +149,6 @@
     "require_jwt",
     "require_jwt_verified",
     "require_scopes",
+    "require_scopes_verified",
     "require_verified_email",
 ]

dependencies/auth.py

@@ -200,6 +200,12 @@ def check_api_key_scope(user: CurrentUser | None, required_scopes: set[str]) ->
     ApiKeyScope.ADMIN_ALL,
 }
 SHORTEN_SCOPES: set[str] = {ApiKeyScope.SHORTEN_CREATE, ApiKeyScope.ADMIN_ALL}
+DOMAIN_MANAGE_SCOPES: set[str] = {ApiKeyScope.DOMAINS_MANAGE, ApiKeyScope.ADMIN_ALL}
+DOMAIN_READ_SCOPES: set[str] = {
+    ApiKeyScope.DOMAINS_MANAGE,
+    ApiKeyScope.DOMAINS_READ,
+    ApiKeyScope.ADMIN_ALL,
+}
 
 
 # ── Parameterised scope dependency factories ─────────────────────────────────
@@ -265,6 +271,24 @@ async def _dep(
     return _dep
 
 
+def require_scopes_verified(scopes: set[str]):
+    """``require_scopes`` plus email-verification gate.
+
+    Use on protected create endpoints that accept BOTH JWT and API key auth.
+    API key callers must also be email-verified (their underlying user must
+    have verified the email at signup time).
+    """
+
+    async def _dep(
+        user: CurrentUser = Depends(require_scopes(scopes)),
+    ) -> CurrentUser:
+        if not user.email_verified:
+            raise EmailNotVerifiedError("Email verification required")
+        return user
+
+    return _dep
+
+
 # ── Annotated type aliases — community-standard Depends shortcuts ─────────────
 
 AuthUser = Annotated[CurrentUser, Depends(require_auth)]

dependencies/wiring.py

@@ -232,4 +232,5 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
         preflight_cname_target=cd_settings.cf_cname_target
         if cd_settings.cf_zone_id
         else None,
+        url_service=app.state.url_service,
     )

infrastructure/cache/url_cache.py

@@ -93,3 +93,19 @@ async def invalidate(self, short_code: str, domain: str) -> None:
             )
         except Exception as e:
             log.error("url_cache_invalidate_error", short_code=short_code, error=str(e))
+
+    async def invalidate_many(self, short_codes: list[str], domain: str) -> None:
+        """Bulk-invalidate cache entries for a list of aliases on one domain."""
+        if not short_codes or self._redis is None:
+            return
+        keys = [self._key(c, domain) for c in short_codes]
+        try:
+            await self._redis.delete(*keys)
+            log.info(
+                "cache_invalidated_bulk",
+                count=len(short_codes),
+                domain=domain,
+                reason="bulk_invalidation",
+            )
+        except Exception as e:
+            log.error("url_cache_invalidate_many_error", domain=domain, error=str(e))

middleware/rate_limiter.py

@@ -73,6 +73,13 @@ class Limits:
     # URL management
     URL_MANAGE = "120 per minute; 2000 per day"
     URL_DELETE = "60 per minute; 1000 per day"
+    URL_BULK_DELETE = "5 per minute; 50 per day"
+
+    # Custom domains
+    DOMAIN_CREATE = "5 per hour"
+    DOMAIN_VERIFY = "10 per minute"
+    DOMAIN_READ = "60 per minute"
+    DOMAIN_DELETE = "10 per minute"
 
     # Dashboard — profile pictures
     PROFILE_PICTURE_SET = "10 per minute"

middleware/tenant.py

@@ -1,18 +1,35 @@
 """Resolve Host header → TenantInfo on every request.
 
 Lands `request.state.tenant` for downstream handlers. Redirect route
-reads it to scope the URL lookup to the right tenant. Unknown public
-hosts get an HTML 404; internal/loopback hosts pass through with
-tenant=None so /health doesn't break.
+reads it to scope the URL lookup to the right tenant.
+
+Routing policy for custom tenants is a strict allowlist:
+  - `GET /<alias>` and `POST /<alias>/password` → redirect router
+  - `GET /favicon.ico`                         → static router (generic favicon)
+  - `GET /robots.txt`                          → inline `Disallow: /`
+  - Everything else                            → 404
+
+Operator surface (`/api/*`, `/dashboard/*`, `/auth/*`, `/oauth/*`, `/health`)
+and brand pages (`/about`, `/contact`, `/api-docs`, `/<alias>+`, `/report`)
+all 404 on custom tenants. Per-domain routing config (`root_redirect`,
+`not_found_redirect`, `custom_robots_txt`) lands in PR4.5.
+
+System-default tenant behaves exactly as before — full app surface.
+
+Every custom-tenant response carries `X-Robots-Tag: noindex, nofollow,
+noarchive` (post-handler). Combined with the disallow-all robots.txt, this
+keeps short links out of search indexes. Preview crawlers (Twitter/Slack/
+Discord/etc.) ignore these signals and continue to unfurl correctly.
 """
 
 from __future__ import annotations
 
+import re
 from urllib.parse import urlsplit
 
 from fastapi import Request
 from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.responses import HTMLResponse, Response
+from starlette.responses import HTMLResponse, PlainTextResponse, Response
 
 from infrastructure.logging import get_logger
 from services.tenant_resolver.protocol import TenantInfo, TenantResolver
@@ -21,6 +38,55 @@
 
 _LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "app"})
 
+_NOT_FOUND_BODY = (
+    "<!doctype html><html><head><title>404 — Not Found</title></head>"
+    "<body><h1>404</h1><p>URL not found.</p></body></html>"
+)
+
+_CUSTOM_TENANT_ROBOTS_BODY = "User-agent: *\nDisallow: /\n"
+_NOINDEX_HEADER = "noindex, nofollow, noarchive"
+
+# Allowed exact paths on custom tenants (besides the alias pattern).
+_ALLOWED_EXACT_PATHS = frozenset({"/favicon.ico"})
+
+# Reserved path prefixes — match these *before* the alias allowlist so
+# operator surface (`/dashboard/*`, `/api/*`, …) and brand pages
+# (`/about`, `/contact`, …) cannot be exposed through the alias namespace.
+# A path is reserved if it equals one of these strings exactly or starts
+# with one followed by `/`. Bare alias collisions (e.g. a customer creating
+# an alias literally named `dashboard`) are sacrificed for tenant isolation.
+_RESERVED_PREFIXES = (
+    "/api",
+    "/dashboard",
+    "/auth",
+    "/oauth",
+    "/health",
+    "/report",
+    "/about",
+    "/contact",
+    "/privacy",
+    "/api-docs",
+    "/api-reference",
+)
+
+# Alias paths allowed on custom tenants. Match `/<alias>` and
+# `/<alias>/password` only. Alias body is `[A-Za-z0-9_-]{3,16}` per
+# `shared.validators.validate_alias` plus the URL-safe slice of the emoji
+# range used in v2. Stats suffix (`+`) is intentionally NOT matched so
+# `/<alias>+` falls through to 404 — analytics surface stays on spoo.me.
+#
+# Emoji ranges: Misc Symbols & Pictographs (1F300-1F5FF), Emoticons
+# (1F600-1F64F), Transport & Map (1F680-1F6FF), Supplemental Symbols
+# (1F900-1F9FF), Extended-A (1FA70-1FAFF), and percent-encoded forms.
+_ALIAS_PATTERN = re.compile(
+    r"^/"
+    r"(?:[A-Za-z0-9_\-]"
+    r"|[\U0001F300-\U0001F5FF\U0001F600-\U0001F64F"
+    r"\U0001F680-\U0001F6FF\U0001F900-\U0001F9FF\U0001FA70-\U0001FAFF]"
+    r"|%[0-9A-Fa-f]{2})+"
+    r"(?:/password)?$"
+)
+
 
 def _normalise_host(raw: str) -> str:
     """Lowercased, dot-stripped, port-stripped host. RFC 3986-safe for
@@ -34,8 +100,30 @@ def _normalise_host(raw: str) -> str:
     return (parsed or "").rstrip(".").lower()
 
 
+def _is_reserved_path(path: str) -> bool:
+    for prefix in _RESERVED_PREFIXES:
+        if path == prefix or path.startswith(prefix + "/"):
+            return True
+    return False
+
+
+def _is_allowed_on_custom_tenant(path: str) -> bool:
+    if path == "/":
+        return False
+    if _is_reserved_path(path):
+        return False
+    if path in _ALLOWED_EXACT_PATHS:
+        return True
+    return bool(_ALIAS_PATTERN.match(path))
+
+
 class TenantMiddleware(BaseHTTPMiddleware):
-    """Populates request.state.tenant from the request Host header."""
+    """Populates request.state.tenant from the request Host header.
+
+    On custom tenants additionally enforces the allowlist routing policy
+    documented at the top of this module and stamps the noindex header on
+    every response.
+    """
 
     async def dispatch(self, request: Request, call_next) -> Response:
         resolver: TenantResolver | None = getattr(
@@ -51,15 +139,35 @@ async def dispatch(self, request: Request, call_next) -> Response:
 
         tenant: TenantInfo | None = await resolver.resolve(host)
         request.state.tenant = tenant
+
         if tenant is None:
             log.info("tenant_unknown_host", host=host)
-            # Static HTML 404 — browser-friendly, no template deps, no
-            # tenancy details leaked.
             return HTMLResponse(_NOT_FOUND_BODY, status_code=404)
-        return await call_next(request)
 
+        if tenant.is_system_default:
+            return await call_next(request)
 
-_NOT_FOUND_BODY = (
-    "<!doctype html><html><head><title>404 — Not Found</title></head>"
-    "<body><h1>404</h1><p>URL not found.</p></body></html>"
-)
+        path = request.url.path
+
+        if path == "/robots.txt":
+            return PlainTextResponse(
+                _CUSTOM_TENANT_ROBOTS_BODY,
+                headers={"X-Robots-Tag": _NOINDEX_HEADER},
+            )
+
+        if not _is_allowed_on_custom_tenant(path):
+            log.info(
+                "tenant_path_denied",
+                host=host,
+                path=path,
+                method=request.method,
+            )
+            return HTMLResponse(
+                _NOT_FOUND_BODY,
+                status_code=404,
+                headers={"X-Robots-Tag": _NOINDEX_HEADER},
+            )
+
+        response = await call_next(request)
+        response.headers["X-Robots-Tag"] = _NOINDEX_HEADER
+        return response

repositories/url_repository.py

@@ -53,6 +53,53 @@ async def delete(self, url_id: ObjectId) -> bool:
         """Hard-delete a URL document. Returns True if a document was deleted."""
         return await self._delete({"_id": url_id})
 
+    async def list_aliases_by_owner_and_domain(
+        self, owner_id: ObjectId, domain: str
+    ) -> list[str]:
+        """Return all aliases owned by *owner_id* under *domain*.
+
+        Used by bulk-delete to drive cache invalidation. Two-step (list then
+        delete) trades atomicity for explicit cache cleanup — a cache miss
+        post-delete is correct behavior anyway.
+        """
+        try:
+            cursor = self._col.find(
+                {"owner_id": owner_id, "domain": domain},
+                projection={"alias": 1, "_id": 0},
+            )
+            docs = await cursor.to_list(length=None)
+            return [d["alias"] for d in docs if "alias" in d]
+        except PyMongoError as exc:
+            log.error(
+                "repo_list_aliases_failed",
+                collection=self._collection_name,
+                error=str(exc),
+            )
+            raise
+
+    async def delete_many_by_owner_and_domain(
+        self, owner_id: ObjectId, domain: str
+    ) -> int:
+        """Bulk-delete all URLs owned by *owner_id* under *domain*.
+
+        Both filters required defensively — a missing or empty arg here would
+        silently delete more than intended.
+        """
+        if not owner_id or not domain:
+            raise ValueError("owner_id and domain are both required for bulk delete")
+        try:
+            result = await self._col.delete_many(
+                {"owner_id": owner_id, "domain": domain}
+            )
+            return int(result.deleted_count or 0)
+        except PyMongoError as exc:
+            log.error(
+                "repo_delete_many_failed",
+                collection=self._collection_name,
+                error=str(exc),
+            )
+            raise
+
     async def check_alias_exists(self, alias: str, domain: str) -> bool:
         """Return True if the alias is taken under the given domain namespace."""
         doc = await self._find_one_raw({"alias": alias, "domain": domain}, {"_id": 1})