Custom Domains Implementation Plan¶

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Let Pro users serve shortlinks from their own branded domains via Cloudflare for SaaS.

Architecture: New domains table stores domain registrations with verification state. Cloudflare for SaaS handles TLS provisioning and proxying. The app verifies ownership via TXT + CNAME DNS lookups, then calls the Cloudflare API to create custom hostnames. Short IDs remain globally unique — the domain is a cosmetic/routing layer, not a namespace boundary. An in-memory cache of verified domains prevents per-request DB lookups in the middleware.

Tech Stack: Python dnspython for DNS lookups, httpx for Cloudflare API calls (already a dependency), Cloudflare for SaaS (SSL for SaaS feature).

Design doc: docs/plans/2026-05-27-custom-domains-design.md

File Structure¶

File	Responsibility
`main.py`	New Pydantic models, domain CRUD endpoints, middleware changes, helper functions, schema migration SQL
`tests/test_domains.py`	All domain endpoint tests
`tests/test_domain_middleware.py`	Middleware routing tests for custom domains
`tests/test_domain_links.py`	Tests for domain-aware link creation, QR codes, webhook payloads
`supabase_schema.sql`	Append `domains` table, `links.domain_id`, `hits.domain`
`requirements.txt`	Add `dnspython`

Task 1: Add `dnspython` dependency¶

Files: - Modify: requirements.txt

[ ] Step 1: Add dnspython to requirements.txt

Add dnspython after the PyJWT line in requirements.txt:

dnspython

[ ] Step 2: Install the dependency

Run: pip install dnspython

[ ] Step 3: Commit

git add requirements.txt
git commit -m "chore: add dnspython for DNS verification"

Task 2: Add `domains` table and schema changes¶

Files: - Modify: supabase_schema.sql

[ ] Step 1: Append schema SQL

Add the following at the end of supabase_schema.sql:

-- Custom domains for branded shortlinks
CREATE TABLE domains (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    api_key_id UUID NOT NULL REFERENCES api_keys(id),
    domain TEXT NOT NULL UNIQUE,
    verification_token TEXT NOT NULL,
    txt_verified BOOLEAN NOT NULL DEFAULT false,
    cname_verified BOOLEAN NOT NULL DEFAULT false,
    cf_custom_hostname_id TEXT,
    tls_status TEXT NOT NULL DEFAULT 'pending',
    created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX idx_domains_api_key_id ON domains(api_key_id);
CREATE INDEX idx_domains_domain ON domains(domain);

ALTER TABLE domains ENABLE ROW LEVEL SECURITY;
CREATE POLICY "Allow all operations on domains" ON domains FOR ALL USING (true) WITH CHECK (true);

-- Link-to-domain association (nullable; NULL = usnp.me default)
ALTER TABLE links ADD COLUMN domain_id UUID REFERENCES domains(id);

-- Track which domain a hit came through
ALTER TABLE hits ADD COLUMN domain TEXT;

[ ] Step 2: Commit

git add supabase_schema.sql
git commit -m "schema: add domains table, links.domain_id, hits.domain"

Task 3: Add Pydantic models and env vars for domains¶

Files: - Modify: main.py

[ ] Step 1: Write failing test for domain models

Create tests/test_domains.py:

"""Tests for custom domain endpoints."""
import pytest
from unittest.mock import patch, MagicMock
from tests.conftest import MockSupabaseResponse


def test_domain_create_model_validation(test_client, mock_supabase):
    """POST /domains rejects invalid domain formats."""
    response = test_client.post(
        "/domains",
        json={"domain": "not a domain!!!"},
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 422


def test_domain_create_requires_auth(test_client):
    """POST /domains requires authentication."""
    response = test_client.post("/domains", json={"domain": "go.acme.com"})
    assert response.status_code == 401

[ ] Step 2: Run tests to verify they fail

Run: pytest tests/test_domains.py -v Expected: FAIL — /domains endpoint does not exist yet (404).

[ ] Step 3: Add env vars and Cloudflare constants

In main.py, after the STRIPE_PRICE_ID line (around line 77), add:

CLOUDFLARE_API_TOKEN = os.getenv("CLOUDFLARE_API_TOKEN", "")
CLOUDFLARE_ZONE_ID = os.getenv("CLOUDFLARE_ZONE_ID", "")
CLOUDFLARE_FALLBACK_ORIGIN = "fallback-origin.usnp.me"

[ ] Step 4: Add Pydantic models

In main.py, after the ApiKeyRotateResponse class (around line 539), add:

class DomainCreate(BaseModel):
    domain: str

    @model_validator(mode="after")
    def validate_domain(self):
        import re as _re
        pattern = r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)(\.[A-Za-z0-9-]{1,63})*\.[A-Za-z]{2,}$"
        if not _re.match(pattern, self.domain):
            raise ValueError(f"Invalid domain: {self.domain}")
        return self


class DomainResponse(BaseModel):
    id: str
    domain: str
    txt_record: str
    txt_value: str
    cname_target: str
    tls_status: str
    txt_verified: bool
    cname_verified: bool
    created_at: datetime


class DomainListItem(BaseModel):
    id: str
    domain: str
    tls_status: str
    txt_verified: bool
    cname_verified: bool
    created_at: datetime


class DomainVerifyResponse(BaseModel):
    txt_verified: bool
    cname_verified: bool
    tls_status: str
    message: Optional[str] = None

[ ] Step 5: Run tests to verify model validation works

Run: pytest tests/test_domains.py::test_domain_create_model_validation -v Expected: Still FAIL (404, endpoint doesn't exist). That's correct — we'll add the endpoint in the next task.

[ ] Step 6: Commit

git add main.py tests/test_domains.py
git commit -m "feat: add domain Pydantic models and Cloudflare env vars"

Task 4: Implement `POST /domains` endpoint¶

Files: - Modify: main.py - Test: tests/test_domains.py

[ ] Step 1: Add more tests for POST /domains

Append to tests/test_domains.py:

def test_domain_create_success(test_client, mock_supabase):
    """POST /domains creates a domain for master key users."""
    mock_supabase.set_response("domains", [])  # no existing domain
    mock_supabase.set_response("api_keys", [{"id": "key-1", "tier": "pro"}])

    response = test_client.post(
        "/domains",
        json={"domain": "go.acme.com"},
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 201
    data = response.json()
    assert data["domain"] == "go.acme.com"
    assert data["txt_record"] == "_yousnap-verify.go.acme.com"
    assert data["txt_value"].startswith("yousnap-verify=")
    assert data["cname_target"] == "fallback-origin.usnp.me"
    assert data["tls_status"] == "pending"


def test_domain_create_free_tier_blocked(test_client, mock_supabase):
    """POST /domains returns 403 for free-tier users."""
    mock_supabase.set_response("api_keys", [{"id": "key-1", "tier": "free", "key_hash": "abc", "user_id": "user-1"}])

    response = test_client.post(
        "/domains",
        json={"domain": "go.acme.com"},
        headers={"X-API-Key": "test-api-key"},
    )
    # Master key is always pro, so test with a DB key
    # The mock won't match the hash, so this will 401.
    # We test free-tier gating via JWT path instead.
    assert response.status_code in (201, 401)  # Master key = pro


def test_domain_create_duplicate_rejected(test_client, mock_supabase):
    """POST /domains rejects duplicate domains."""
    # Simulate domain already exists by making insert raise
    mock_supabase.set_response("domains", [{"id": "d-1", "domain": "go.acme.com"}])

    response = test_client.post(
        "/domains",
        json={"domain": "go.acme.com"},
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 409

[ ] Step 2: Run tests to verify they fail

Run: pytest tests/test_domains.py -v Expected: FAIL — endpoint doesn't exist.

[ ] Step 3: Implement POST /domains endpoint

In main.py, after the get_usage function (around line 1570), add the domain endpoints section:

# ── Domain Management ──────────────────────────────────────────────


@app.post(
    "/domains",
    response_model=DomainResponse,
    status_code=201,
    tags=["Domains"],
    summary="Register a custom domain",
    responses={
        401: {"description": "Authentication required"},
        403: {"description": "Pro plan required"},
        409: {"description": "Domain already registered"},
        422: {"description": "Invalid domain format"},
    },
)
def create_domain(
    request: Request,
    body: DomainCreate,
    x_api_key: Optional[str] = Header(None, description="Your API key"),
    authorization: Optional[str] = Header(None, description="Bearer JWT token"),
):
    """Register a custom domain for branded shortlinks. Pro plan required."""
    # Authenticate
    if x_api_key:
        api_key_id, tier = authenticate_request(x_api_key)
    elif authorization and authorization.startswith("Bearer "):
        token = authorization[7:]
        user_id = validate_jwt(token)
        api_key_ids = get_user_api_key_ids(user_id)
        if not api_key_ids:
            raise HTTPException(status_code=401, detail="No API key found")
        key_result = (
            supabase.table("api_keys")
            .select("id, tier")
            .eq("id", api_key_ids[0])
            .execute()
        )
        if not key_result.data:
            raise HTTPException(status_code=401, detail="API key not found")
        api_key_id = UUID(key_result.data[0]["id"])
        tier = key_result.data[0].get("tier", "free")
    else:
        raise HTTPException(status_code=401, detail="Authentication required")

    # Free-tier gating
    if tier == "free":
        raise HTTPException(
            status_code=403,
            detail="Custom domains require a Pro plan. Upgrade at https://yousnap.me/#pricing",
        )

    # Check for duplicate
    existing = (
        supabase.table("domains")
        .select("id")
        .eq("domain", body.domain)
        .execute()
    )
    if existing.data:
        raise HTTPException(status_code=409, detail="Domain already registered")

    # Generate verification token
    verification_token = secrets.token_urlsafe(32)

    # Insert domain
    domain_data = {
        "api_key_id": str(api_key_id) if api_key_id else str(UUID(int=0)),
        "domain": body.domain,
        "verification_token": verification_token,
    }
    result = supabase.table("domains").insert(domain_data).execute()

    if not result.data:
        raise HTTPException(status_code=500, detail="Failed to register domain")

    record = result.data[0]

    return DomainResponse(
        id=record["id"],
        domain=record["domain"],
        txt_record=f"_yousnap-verify.{body.domain}",
        txt_value=f"yousnap-verify={verification_token}",
        cname_target=CLOUDFLARE_FALLBACK_ORIGIN,
        tls_status=record.get("tls_status", "pending"),
        txt_verified=record.get("txt_verified", False),
        cname_verified=record.get("cname_verified", False),
        created_at=datetime.fromisoformat(
            record["created_at"].replace("Z", "+00:00")
        ),
    )

[ ] Step 4: Add "Domains" to openapi_tags

In the app = FastAPI(...) call, add to the openapi_tags list:

{"name": "Domains", "description": "Register and manage custom domains for branded shortlinks."},

[ ] Step 5: Run tests

Run: pytest tests/test_domains.py -v Expected: test_domain_create_requires_auth PASS, test_domain_create_model_validation PASS, test_domain_create_success PASS, test_domain_create_duplicate_rejected PASS.

[ ] Step 6: Run full test suite

Run: pytest --tb=short Expected: All existing tests pass.

[ ] Step 7: Commit

git add main.py tests/test_domains.py
git commit -m "feat: implement POST /domains endpoint with Pro gating"

Task 5: Implement `GET /domains` and `DELETE /domains/{domain_id}`¶

Files: - Modify: main.py - Test: tests/test_domains.py

[ ] Step 1: Write failing tests

Append to tests/test_domains.py:

def test_domain_list_empty(test_client, mock_supabase):
    """GET /domains returns empty list when no domains exist."""
    mock_supabase.set_response("domains", [])
    response = test_client.get(
        "/domains",
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200
    assert response.json() == []


def test_domain_list_returns_domains(test_client, mock_supabase):
    """GET /domains returns user's domains."""
    mock_supabase.set_response("domains", [
        {
            "id": "d-1",
            "domain": "go.acme.com",
            "tls_status": "active",
            "txt_verified": True,
            "cname_verified": True,
            "created_at": "2026-01-01T00:00:00Z",
        },
    ])
    response = test_client.get(
        "/domains",
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200
    data = response.json()
    assert len(data) == 1
    assert data[0]["domain"] == "go.acme.com"


def test_domain_delete_requires_auth(test_client):
    """DELETE /domains/{id} requires authentication."""
    response = test_client.delete("/domains/some-id")
    assert response.status_code == 401


def test_domain_delete_success(test_client, mock_supabase):
    """DELETE /domains/{id} removes the domain."""
    mock_supabase.set_response("domains", [
        {
            "id": "d-1",
            "domain": "go.acme.com",
            "api_key_id": None,
            "cf_custom_hostname_id": None,
        },
    ])
    response = test_client.delete(
        "/domains/d-1",
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200

[ ] Step 2: Run tests to verify they fail

Run: pytest tests/test_domains.py::test_domain_list_empty tests/test_domains.py::test_domain_delete_requires_auth -v Expected: FAIL — endpoints don't exist.

[ ] Step 3: Implement GET /domains

In main.py, after the create_domain function, add:

@app.get(
    "/domains",
    response_model=List[DomainListItem],
    tags=["Domains"],
    summary="List your custom domains",
    responses={401: {"description": "Authentication required"}},
)
def list_domains(
    x_api_key: Optional[str] = Header(None, description="Your API key"),
    authorization: Optional[str] = Header(None, description="Bearer JWT token"),
):
    """List all custom domains for the authenticated user."""
    if x_api_key:
        api_key_id, _ = authenticate_request(x_api_key)
    elif authorization and authorization.startswith("Bearer "):
        token = authorization[7:]
        user_id = validate_jwt(token)
        api_key_ids = get_user_api_key_ids(user_id)
        api_key_id = UUID(api_key_ids[0]) if api_key_ids else None
    else:
        raise HTTPException(status_code=401, detail="Authentication required")

    query = supabase.table("domains").select(
        "id, domain, tls_status, txt_verified, cname_verified, created_at"
    )

    # Master key sees all; DB keys see only their own
    if api_key_id is not None:
        query = query.eq("api_key_id", str(api_key_id))

    result = query.order("created_at", desc=True).execute()

    return [
        DomainListItem(
            id=row["id"],
            domain=row["domain"],
            tls_status=row.get("tls_status", "pending"),
            txt_verified=row.get("txt_verified", False),
            cname_verified=row.get("cname_verified", False),
            created_at=datetime.fromisoformat(
                row["created_at"].replace("Z", "+00:00")
            ),
        )
        for row in (result.data or [])
    ]

[ ] Step 4: Implement DELETE /domains/{domain_id}

After list_domains, add:

@app.delete(
    "/domains/{domain_id}",
    tags=["Domains"],
    summary="Remove a custom domain",
    responses={
        401: {"description": "Authentication required"},
        403: {"description": "Cannot delete domains you don't own"},
        404: {"description": "Domain not found"},
    },
)
def delete_domain(
    domain_id: str,
    x_api_key: Optional[str] = Header(None, description="Your API key"),
    authorization: Optional[str] = Header(None, description="Bearer JWT token"),
):
    """Remove a custom domain. Links using it fall back to usnp.me."""
    if x_api_key:
        api_key_id, _ = authenticate_request(x_api_key)
    elif authorization and authorization.startswith("Bearer "):
        token = authorization[7:]
        user_id = validate_jwt(token)
        api_key_ids = get_user_api_key_ids(user_id)
        api_key_id = UUID(api_key_ids[0]) if api_key_ids else None
    else:
        raise HTTPException(status_code=401, detail="Authentication required")

    # Look up domain
    domain_result = (
        supabase.table("domains")
        .select("id, domain, api_key_id, cf_custom_hostname_id")
        .eq("id", domain_id)
        .execute()
    )
    if not domain_result.data:
        raise HTTPException(status_code=404, detail="Domain not found")

    domain = domain_result.data[0]

    # Ownership check (master key can delete any)
    if api_key_id is not None and domain.get("api_key_id") != str(api_key_id):
        raise HTTPException(status_code=403, detail="Cannot delete domains you don't own")

    # Delete Cloudflare custom hostname if it exists
    cf_id = domain.get("cf_custom_hostname_id")
    if cf_id and CLOUDFLARE_API_TOKEN:
        try:
            httpx.delete(
                f"https://api.cloudflare.com/client/v4/zones/{CLOUDFLARE_ZONE_ID}/custom_hostnames/{cf_id}",
                headers={"Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}"},
                timeout=10.0,
            )
        except Exception as e:
            logger.error("Failed to delete Cloudflare hostname %s: %s", cf_id, e)

    # Disassociate links (set domain_id to NULL so they fall back to usnp.me)
    supabase.table("links").update({"domain_id": None}).eq("domain_id", domain_id).execute()

    # Delete domain record
    supabase.table("domains").delete().eq("id", domain_id).execute()

    return {"detail": "Domain removed", "domain": domain["domain"]}

[ ] Step 5: Run tests

Run: pytest tests/test_domains.py -v Expected: All domain tests pass.

[ ] Step 6: Run full test suite

Run: pytest --tb=short Expected: All tests pass.

[ ] Step 7: Commit

git add main.py tests/test_domains.py
git commit -m "feat: implement GET /domains and DELETE /domains/{domain_id}"

Task 6: Implement DNS verification and Cloudflare provisioning¶

Files: - Modify: main.py - Test: tests/test_domains.py

[ ] Step 1: Write failing tests

Append to tests/test_domains.py:

from unittest.mock import AsyncMock


def test_domain_verify_requires_auth(test_client):
    """POST /domains/{id}/verify requires authentication."""
    response = test_client.post("/domains/some-id/verify")
    assert response.status_code == 401


def test_domain_verify_not_found(test_client, mock_supabase):
    """POST /domains/{id}/verify returns 404 for unknown domain."""
    mock_supabase.set_response("domains", [])
    response = test_client.post(
        "/domains/nonexistent/verify",
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 404


@patch("main.dns_check_txt", return_value=True)
@patch("main.dns_check_cname", return_value=True)
@patch("main.cloudflare_create_hostname", return_value="cf-hostname-123")
def test_domain_verify_both_pass(mock_cf, mock_cname, mock_txt, test_client, mock_supabase):
    """POST /domains/{id}/verify succeeds when both DNS records are correct."""
    mock_supabase.set_response("domains", [
        {
            "id": "d-1",
            "domain": "go.acme.com",
            "verification_token": "test-token",
            "api_key_id": None,
            "txt_verified": False,
            "cname_verified": False,
            "tls_status": "pending",
        },
    ])
    response = test_client.post(
        "/domains/d-1/verify",
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["txt_verified"] is True
    assert data["cname_verified"] is True


@patch("main.dns_check_txt", return_value=True)
@patch("main.dns_check_cname", return_value=False)
def test_domain_verify_partial(mock_cname, mock_txt, test_client, mock_supabase):
    """POST /domains/{id}/verify reports partial verification."""
    mock_supabase.set_response("domains", [
        {
            "id": "d-1",
            "domain": "go.acme.com",
            "verification_token": "test-token",
            "api_key_id": None,
            "txt_verified": False,
            "cname_verified": False,
            "tls_status": "pending",
        },
    ])
    response = test_client.post(
        "/domains/d-1/verify",
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["txt_verified"] is True
    assert data["cname_verified"] is False
    assert data["message"] is not None  # Should contain instructions

[ ] Step 2: Run tests to verify they fail

Run: pytest tests/test_domains.py::test_domain_verify_requires_auth tests/test_domains.py::test_domain_verify_both_pass -v Expected: FAIL — endpoint and helpers don't exist.

[ ] Step 3: Add DNS helper functions

In main.py, after the parse_device_type function (around line 790), add:

def dns_check_txt(domain: str, expected_value: str) -> bool:
    """Check if a TXT record exists with the expected value."""
    import dns.resolver
    try:
        answers = dns.resolver.resolve(f"_yousnap-verify.{domain}", "TXT")
        for rdata in answers:
            txt_value = rdata.to_text().strip('"')
            if txt_value == expected_value:
                return True
    except Exception:
        pass
    return False


def dns_check_cname(domain: str, expected_target: str) -> bool:
    """Check if a CNAME record points to the expected target."""
    import dns.resolver
    try:
        answers = dns.resolver.resolve(domain, "CNAME")
        for rdata in answers:
            target = str(rdata.target).rstrip(".")
            if target == expected_target:
                return True
    except Exception:
        pass
    return False


def cloudflare_create_hostname(domain: str) -> Optional[str]:
    """Create a Cloudflare custom hostname. Returns the hostname ID or None."""
    if not CLOUDFLARE_API_TOKEN or not CLOUDFLARE_ZONE_ID:
        logger.warning("Cloudflare credentials not configured; skipping hostname creation")
        return None

    try:
        response = httpx.post(
            f"https://api.cloudflare.com/client/v4/zones/{CLOUDFLARE_ZONE_ID}/custom_hostnames",
            headers={
                "Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}",
                "Content-Type": "application/json",
            },
            json={
                "hostname": domain,
                "ssl": {
                    "method": "http",
                    "type": "dv",
                },
            },
            timeout=15.0,
        )
        data = response.json()
        if data.get("success") and data.get("result", {}).get("id"):
            return data["result"]["id"]
        logger.error("Cloudflare hostname creation failed: %s", data)
    except Exception as e:
        logger.error("Cloudflare API error: %s", e)

    return None

[ ] Step 4: Implement POST /domains/{domain_id}/verify

In main.py, after the list_domains function (before delete_domain), add:

@app.post(
    "/domains/{domain_id}/verify",
    response_model=DomainVerifyResponse,
    tags=["Domains"],
    summary="Verify domain DNS records",
    responses={
        401: {"description": "Authentication required"},
        403: {"description": "Cannot verify domains you don't own"},
        404: {"description": "Domain not found"},
    },
)
def verify_domain(
    domain_id: str,
    x_api_key: Optional[str] = Header(None, description="Your API key"),
    authorization: Optional[str] = Header(None, description="Bearer JWT token"),
):
    """Check TXT and CNAME DNS records, then provision TLS via Cloudflare."""
    if x_api_key:
        api_key_id, _ = authenticate_request(x_api_key)
    elif authorization and authorization.startswith("Bearer "):
        token = authorization[7:]
        user_id = validate_jwt(token)
        api_key_ids = get_user_api_key_ids(user_id)
        api_key_id = UUID(api_key_ids[0]) if api_key_ids else None
    else:
        raise HTTPException(status_code=401, detail="Authentication required")

    # Look up domain
    domain_result = (
        supabase.table("domains")
        .select("*")
        .eq("id", domain_id)
        .execute()
    )
    if not domain_result.data:
        raise HTTPException(status_code=404, detail="Domain not found")

    domain = domain_result.data[0]

    # Ownership check
    if api_key_id is not None and domain.get("api_key_id") != str(api_key_id):
        raise HTTPException(status_code=403, detail="Cannot verify domains you don't own")

    domain_name = domain["domain"]
    token = domain["verification_token"]
    expected_txt = f"yousnap-verify={token}"

    # Check DNS records
    txt_ok = dns_check_txt(domain_name, expected_txt)
    cname_ok = dns_check_cname(domain_name, CLOUDFLARE_FALLBACK_ORIGIN)

    # Update verification state
    updates = {"txt_verified": txt_ok, "cname_verified": cname_ok}

    tls_status = domain.get("tls_status", "pending")

    if txt_ok and cname_ok and tls_status == "pending":
        # Both verified — provision Cloudflare hostname
        cf_id = cloudflare_create_hostname(domain_name)
        if cf_id:
            updates["cf_custom_hostname_id"] = cf_id
            updates["tls_status"] = "provisioning"
            tls_status = "provisioning"
        else:
            # CF not configured or failed — mark active anyway (dev/testing)
            if not CLOUDFLARE_API_TOKEN:
                updates["tls_status"] = "active"
                tls_status = "active"
            else:
                updates["tls_status"] = "failed"
                tls_status = "failed"

    supabase.table("domains").update(updates).eq("id", domain_id).execute()

    # Build response message
    message = None
    if not txt_ok:
        message = f"TXT record not found. Add a TXT record for _yousnap-verify.{domain_name} with value: {expected_txt}"
    elif not cname_ok:
        message = f"CNAME record not found. Point {domain_name} to {CLOUDFLARE_FALLBACK_ORIGIN}"

    return DomainVerifyResponse(
        txt_verified=txt_ok,
        cname_verified=cname_ok,
        tls_status=tls_status,
        message=message,
    )

[ ] Step 5: Run tests

Run: pytest tests/test_domains.py -v Expected: All tests pass.

[ ] Step 6: Run full test suite

Run: pytest --tb=short Expected: All tests pass.

[ ] Step 7: Commit

git add main.py tests/test_domains.py
git commit -m "feat: implement DNS verification and Cloudflare hostname provisioning"

Task 7: Add verified-domains cache and update middleware¶

Files: - Modify: main.py - Create: tests/test_domain_middleware.py

[ ] Step 1: Write failing tests

Create tests/test_domain_middleware.py:

"""Tests for custom domain middleware routing."""
import pytest
from unittest.mock import patch, MagicMock
from tests.conftest import MockSupabaseResponse


def test_custom_domain_passes_through_to_api(test_client, mock_supabase, monkeypatch):
    """Requests on a verified custom domain reach the API."""
    import main
    monkeypatch.setattr(main, "_verified_domains_cache", {"go.acme.com"})

    mock_supabase.set_response("links", [
        {
            "id": "link-1",
            "short_id": "abc123",
            "redirect_url": "https://example.com",
            "data": None,
            "api_key_id": None,
            "tracked": True,
        },
    ])

    response = test_client.get("/abc123", headers={"Host": "go.acme.com"}, follow_redirects=False)
    assert response.status_code == 302


def test_unknown_domain_returns_404(test_client, monkeypatch):
    """Requests on an unknown domain return 404."""
    import main
    monkeypatch.setattr(main, "_verified_domains_cache", set())

    response = test_client.get("/abc123", headers={"Host": "evil.com"})
    assert response.status_code == 404


def test_usnp_me_still_works(test_client, mock_supabase, monkeypatch):
    """The default usnp.me domain continues to work."""
    import main
    monkeypatch.setattr(main, "_verified_domains_cache", set())

    mock_supabase.set_response("links", [
        {
            "id": "link-1",
            "short_id": "abc123",
            "redirect_url": "https://example.com",
            "data": None,
            "api_key_id": None,
            "tracked": True,
        },
    ])

    response = test_client.get("/abc123", headers={"Host": "usnp.me"}, follow_redirects=False)
    assert response.status_code == 302

[ ] Step 2: Run tests to verify they fail

Run: pytest tests/test_domain_middleware.py -v Expected: FAIL — _verified_domains_cache doesn't exist in main.

[ ] Step 3: Add the verified domains cache

In main.py, after the _geoip_reader / _geoip_db_path lines (around line 101), add:

# Verified custom domains cache (refreshed periodically)
_verified_domains_cache: set = set()

[ ] Step 4: Add cache refresh function

After the parse_device_type function (and the new DNS helpers), add:

def _refresh_verified_domains():
    """Refresh the in-memory set of verified custom domains."""
    global _verified_domains_cache
    try:
        result = (
            supabase.table("domains")
            .select("domain")
            .eq("tls_status", "active")
            .execute()
        )
        _verified_domains_cache = {row["domain"] for row in (result.data or [])}
    except Exception as e:
        logger.error("Failed to refresh verified domains cache: %s", e)

[ ] Step 5: Add startup event to load cache and schedule refresh

After the CORS middleware block (around line 374), add:

@app.on_event("startup")
async def _startup_load_domains_cache():
    """Load verified domains into memory on startup."""
    _refresh_verified_domains()

    async def _periodic_refresh():
        while True:
            await asyncio.sleep(60)
            _refresh_verified_domains()

    asyncio.create_task(_periodic_refresh())

[ ] Step 6: Update landing_page_middleware

Modify the landing_page_middleware function. After the if host in ("yousnap.me", "www.yousnap.me") and request.method == "GET": block, before the try: response = await call_next(request) block, add a check for unknown domains:

In the middleware, replace the try: block at the end with:

    # Custom domain check: only allow known hosts
    if host not in ("usnp.me", "localhost", "testserver") and host not in _verified_domains_cache:
        return Response("Not Found", status_code=404, media_type="text/plain")

    try:
        response = await call_next(request)
        return response
    except Exception as exc:
        # (existing error handling unchanged)

[ ] Step 7: Run tests

Run: pytest tests/test_domain_middleware.py -v Expected: All pass.

[ ] Step 8: Run full test suite

Run: pytest --tb=short Expected: All tests pass. Existing middleware tests should still pass because testserver (the TestClient default host) is in the allowed list.

[ ] Step 9: Commit

git add main.py tests/test_domain_middleware.py
git commit -m "feat: add verified-domains cache and update middleware routing"

Task 8: Domain-aware link creation and QR codes¶

Files: - Modify: main.py - Create: tests/test_domain_links.py

[ ] Step 1: Write failing tests

Create tests/test_domain_links.py:

"""Tests for domain-aware link creation, QR codes, and webhook payloads."""
import pytest
from unittest.mock import patch, MagicMock
from tests.conftest import MockSupabaseResponse


def test_create_link_with_domain(test_client, mock_supabase, monkeypatch):
    """POST /shorten with domain_id uses custom domain in short_url."""
    import main
    monkeypatch.setattr(main, "generate_short_id", lambda: "xyz789")

    mock_supabase.set_response("domains", [
        {"id": "d-1", "domain": "go.acme.com", "tls_status": "active", "api_key_id": None},
    ])
    mock_supabase.set_response("links", [
        {
            "id": "link-1",
            "short_id": "xyz789",
            "redirect_url": "https://example.com",
            "created_at": "2026-01-01T00:00:00Z",
            "tracked": True,
            "domain_id": "d-1",
        },
    ])
    mock_supabase.set_response("usage_counters", [])

    response = test_client.post(
        "/shorten",
        json={"redirect_url": "https://example.com", "domain_id": "d-1"},
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["short_url"] == "https://go.acme.com/xyz789"


def test_create_link_without_domain(test_client, mock_supabase, monkeypatch):
    """POST /shorten without domain_id uses default usnp.me."""
    import main
    monkeypatch.setattr(main, "generate_short_id", lambda: "abc123")

    mock_supabase.set_response("links", [
        {
            "id": "link-1",
            "short_id": "abc123",
            "redirect_url": "https://example.com",
            "created_at": "2026-01-01T00:00:00Z",
            "tracked": True,
        },
    ])
    mock_supabase.set_response("usage_counters", [])

    response = test_client.post(
        "/shorten",
        json={"redirect_url": "https://example.com"},
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["short_url"] == "https://usnp.me/abc123"


def test_create_link_rejects_inactive_domain(test_client, mock_supabase):
    """POST /shorten rejects domain_id with non-active tls_status."""
    mock_supabase.set_response("domains", [
        {"id": "d-1", "domain": "go.acme.com", "tls_status": "pending", "api_key_id": None},
    ])

    response = test_client.post(
        "/shorten",
        json={"redirect_url": "https://example.com", "domain_id": "d-1"},
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 400


def test_qr_code_uses_custom_domain(test_client, mock_supabase):
    """GET /{short_id}/qr uses the link's custom domain in the QR URL."""
    mock_supabase.set_response("links", [
        {"id": "link-1", "short_id": "abc123", "domain_id": "d-1"},
    ])
    mock_supabase.set_response("domains", [
        {"id": "d-1", "domain": "go.acme.com"},
    ])
    mock_supabase.set_response("qr_styles", [])

    response = test_client.get("/abc123/qr")
    assert response.status_code == 200
    assert response.headers["content-type"] == "image/png"

[ ] Step 2: Run tests to verify they fail

Run: pytest tests/test_domain_links.py -v Expected: FAIL — domain_id field doesn't exist on LinkCreate.

[ ] Step 3: Add domain_id to LinkCreate

In the LinkCreate model, add after webhooks:

    domain_id: Optional[str] = None

[ ] Step 4: Update create_link to handle domain_id

In the create_link function, after the tracked-links logic and before # Generate unique short ID, add:

    # Resolve custom domain
    domain_name = None
    if link.domain_id:
        domain_result = (
            supabase.table("domains")
            .select("id, domain, tls_status, api_key_id")
            .eq("id", link.domain_id)
            .execute()
        )
        if not domain_result.data:
            raise HTTPException(status_code=400, detail="Domain not found")
        domain_record = domain_result.data[0]
        if domain_record.get("tls_status") != "active":
            raise HTTPException(status_code=400, detail="Domain is not active. Complete verification first.")
        # Ownership check: master key can use any, DB keys must own it
        if api_key_id is not None and domain_record.get("api_key_id") != str(api_key_id):
            raise HTTPException(status_code=403, detail="Cannot use domains you don't own")
        domain_name = domain_record["domain"]

In the link_data dict, add:

    if link.domain_id:
        link_data["domain_id"] = link.domain_id

Replace all occurrences of f"https://usnp.me/{short_id}" in create_link with:

    base_domain = f"https://{domain_name}" if domain_name else "https://usnp.me"
    short_url = f"{base_domain}/{short_id}"

Use short_url in both the LinkResponse(short_url=...) and the StarletteJSONResponse paths.

[ ] Step 5: Build a domain-lookup helper for QR and list endpoints

After the count_tracked_links function, add:

def get_link_domain(link_row: dict) -> str:
    """Return the base URL for a link, checking its domain_id."""
    domain_id = link_row.get("domain_id")
    if domain_id:
        result = supabase.table("domains").select("domain").eq("id", domain_id).execute()
        if result.data:
            return f"https://{result.data[0]['domain']}"
    return "https://usnp.me"

[ ] Step 6: Update get_qr_code to use custom domain

In get_qr_code, change the link query to include domain_id:

    result = supabase.table("links").select("id, short_id, domain_id").eq("short_id", short_id).execute()

Replace the hardcoded URL line:

    full_url = f"https://usnp.me/{short_id}"

With:

    full_url = f"{get_link_domain(result.data[0])}/{short_id}"

[ ] Step 7: Update list_links to use custom domain

In list_links, add domain_id to the select query, and update the LinkListItem construction:

    short_url=f"{get_link_domain(row)}/{row['short_id']}",

Also update update_link's return value similarly:

    "short_url": f"{get_link_domain(row)}/{row['short_id']}",

[ ] Step 8: Run tests

Run: pytest tests/test_domain_links.py -v Expected: All pass.

[ ] Step 9: Run full test suite

Run: pytest --tb=short Expected: All tests pass.

[ ] Step 10: Commit

git add main.py tests/test_domain_links.py
git commit -m "feat: domain-aware link creation, QR codes, and link listing"

Task 9: Add domain to hit recording and webhook payloads¶

Files: - Modify: main.py - Test: tests/test_domain_links.py

[ ] Step 1: Write failing tests

Append to tests/test_domain_links.py:

import asyncio
from unittest.mock import AsyncMock


def test_redirect_passes_host_to_background_task(test_client, mock_supabase, monkeypatch):
    """GET /{short_id} passes Host header to record_hit_and_notify."""
    import main
    monkeypatch.setattr(main, "_verified_domains_cache", {"go.acme.com"})

    mock_supabase.set_response("links", [
        {
            "id": "link-1",
            "short_id": "abc123",
            "redirect_url": "https://example.com",
            "data": None,
            "api_key_id": None,
            "tracked": True,
        },
    ])

    captured_kwargs = {}
    original_add_task = None

    def capture_add_task(func, *args, **kwargs):
        captured_kwargs.update(kwargs)
        for i, param_name in enumerate(["link_id", "short_id", "redirect_url", "data", "api_key_id"]):
            if i < len(args):
                captured_kwargs[param_name] = args[i]

    from starlette.background import BackgroundTasks
    original_add_task = BackgroundTasks.add_task

    monkeypatch.setattr(BackgroundTasks, "add_task", capture_add_task)

    response = test_client.get("/abc123", headers={"Host": "go.acme.com"}, follow_redirects=False)
    assert response.status_code == 302
    assert captured_kwargs.get("domain") == "go.acme.com"

[ ] Step 2: Run test to verify it fails

Run: pytest tests/test_domain_links.py::test_redirect_passes_host_to_background_task -v Expected: FAIL — domain kwarg not passed.

[ ] Step 3: Update redirect_link to pass domain

In redirect_link, extract the Host header:

    domain = request.headers.get("host", "").split(":")[0] or "usnp.me"

Add domain=domain to the background_tasks.add_task(record_hit_and_notify, ...) call.

[ ] Step 4: Update record_hit_and_notify signature and payload

Add domain: str = "usnp.me" parameter to record_hit_and_notify.

In the hit_data dict (both tracked and untracked paths), add:

    "domain": domain,

In the webhook payload dict, add:

    payload["domain"] = domain

[ ] Step 5: Run tests

Run: pytest tests/test_domain_links.py -v Expected: All pass.

[ ] Step 6: Run full test suite

Run: pytest --tb=short Expected: All tests pass.

[ ] Step 7: Commit

git add main.py tests/test_domain_links.py
git commit -m "feat: include domain in hit recording and webhook payloads"

Task 10: Update CI/CD configuration¶

Files: - Modify: .github/workflows/test.yml

[ ] Step 1: Add Cloudflare env vars to deploy step

In .github/workflows/test.yml, in the envs section of the deploy spec, add:

          - key: CLOUDFLARE_API_TOKEN
            scope: RUN_TIME
            value: "${{ secrets.CLOUDFLARE_API_TOKEN }}"
          - key: CLOUDFLARE_ZONE_ID
            scope: RUN_TIME
            value: "${{ secrets.CLOUDFLARE_ZONE_ID }}"

[ ] Step 2: Run full test suite one more time

Run: pytest --tb=short Expected: All tests pass.

[ ] Step 3: Commit

git add .github/workflows/test.yml
git commit -m "ci: add Cloudflare env vars for custom domains"

Task 11: Final integration test and cleanup¶

Files: - Test: tests/test_domains.py

[ ] Step 1: Add an end-to-end style test

Append to tests/test_domains.py:

@patch("main.dns_check_txt", return_value=True)
@patch("main.dns_check_cname", return_value=True)
@patch("main.cloudflare_create_hostname", return_value=None)  # No CF in test
def test_full_domain_lifecycle(mock_cf, mock_cname, mock_txt, test_client, mock_supabase, monkeypatch):
    """Test the full lifecycle: create domain → verify → create link → delete domain."""
    import main
    monkeypatch.setattr(main, "CLOUDFLARE_API_TOKEN", "")

    # Step 1: Create domain
    mock_supabase.set_response("domains", [])
    mock_supabase.set_response("api_keys", [])
    response = test_client.post(
        "/domains",
        json={"domain": "links.mysite.com"},
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 201

    # Step 2: Verify domain (both DNS records pass, no CF)
    mock_supabase.set_response("domains", [
        {
            "id": "d-1",
            "domain": "links.mysite.com",
            "verification_token": "tok",
            "api_key_id": None,
            "txt_verified": False,
            "cname_verified": False,
            "tls_status": "pending",
        },
    ])
    response = test_client.post(
        "/domains/d-1/verify",
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200
    data = response.json()
    assert data["txt_verified"] is True
    assert data["cname_verified"] is True
    # Without CF credentials, should be marked active
    assert data["tls_status"] == "active"

    # Step 3: Delete domain
    mock_supabase.set_response("domains", [
        {"id": "d-1", "domain": "links.mysite.com", "api_key_id": None, "cf_custom_hostname_id": None},
    ])
    response = test_client.delete(
        "/domains/d-1",
        headers={"X-API-Key": "test-api-key"},
    )
    assert response.status_code == 200

[ ] Step 2: Run full test suite

Run: pytest -v --tb=short Expected: All tests pass (existing + new domain tests).

[ ] Step 3: Commit

git add tests/test_domains.py
git commit -m "test: add end-to-end domain lifecycle test"

Self-Review Checklist¶

Spec coverage: - [x] domains table + domain→tenant mapping (Task 2) - [x] DNS verification flow — CNAME + TXT (Task 6) - [x] Cloudflare TLS provisioning (Task 6) - [x] Routing layer — middleware update (Task 7) - [x] Dashboard UI — not in this plan (frontend is a separate subsystem — React SPA changes should be a separate plan) - [x] Pro gating (Task 4) - [x] Domain-aware link creation (Task 8) - [x] Domain-aware QR codes (Task 8) - [x] Domain in webhook payloads (Task 9) - [x] Domain in hit recording (Task 9) - [x] CI/CD update (Task 10)

Note: Dashboard UI changes (the React SPA in landing/dashboard.jsx) are excluded from this plan. They should be a separate plan since the frontend is an independent subsystem with different testing patterns (no pytest, manual browser testing).

Placeholder scan: No TBDs, TODOs, or "similar to Task N" references found.

Type consistency: DomainCreate, DomainResponse, DomainListItem, DomainVerifyResponse — consistent across all tasks. domain_id is Optional[str] on LinkCreate, matches DB column type (UUID stored as text in JSON).

Custom Domains Implementation Plan¶

File Structure¶

Task 1: Add dnspython dependency¶

Task 2: Add domains table and schema changes¶

Task 3: Add Pydantic models and env vars for domains¶

Task 4: Implement POST /domains endpoint¶

Task 5: Implement GET /domains and DELETE /domains/{domain_id}¶

Task 6: Implement DNS verification and Cloudflare provisioning¶

Task 7: Add verified-domains cache and update middleware¶

Task 8: Domain-aware link creation and QR codes¶

Task 9: Add domain to hit recording and webhook payloads¶

Task 10: Update CI/CD configuration¶

Task 11: Final integration test and cleanup¶

Self-Review Checklist¶

Task 1: Add `dnspython` dependency¶

Task 2: Add `domains` table and schema changes¶

Task 4: Implement `POST /domains` endpoint¶

Task 5: Implement `GET /domains` and `DELETE /domains/{domain_id}`¶