Tenant Domain Onboarding — Build Domain Setup Flows for Your Customers

from ark import Ark

client = Ark()

def onboard_customer_domain(tenant_id: str, domain_name: str):
    """Create a sending domain for a customer tenant."""
    domain = client.domains.create(
        name=domain_name,
        tenant_id=tenant_id
    )

    return {
        "domain_id": domain.data.id,
        "dns_records": domain.data.dns_records,
        "status": domain.data.verification_status
    }

# Usage
result = onboard_customer_domain("tenant_abc123", "mail.customerdomain.com")
print(f"Domain created: {result['domain_id']}")
print(f"DNS records to configure: {len(result['dns_records'])}")

def format_dns_instructions(domain_data):
    """Format DNS records for display in your customer UI."""
    records = domain_data["dns_records"]
    instructions = []

    for record in records:
        instructions.append({
            "type": record["type"],          # TXT, CNAME, MX
            "hostname": record["hostname"],  # e.g., "ark._domainkey.customerdomain.com"
            "value": record["value"],        # The record value to set
            "purpose": record["purpose"],    # "spf", "dkim", "return_path"
            "verified": record["verified"]   # True/False
        })

    return instructions

# Example output for your customer UI:
# ┌──────────┬───────────────────────────────────┬──────────────────┐
# │ Type     │ Hostname                          │ Purpose          │
# ├──────────┼───────────────────────────────────┼──────────────────┤
# │ TXT      │ customerdomain.com                │ SPF              │
# │ TXT      │ ark._domainkey.customerdomain.com │ DKIM             │
# │ CNAME    │ rp.customerdomain.com             │ Return Path      │
# └──────────┴───────────────────────────────────┴──────────────────┘

import time
import ark

def verify_domain_with_polling(domain_id: str, max_attempts: int = 10):
    """
    Poll domain verification with exponential backoff.
    Returns verification result or raises after max attempts.
    """
    for attempt in range(max_attempts):
        result = client.domains.verify(domain_id)
        status = result.data.verification_status

        if status == "verified":
            return {"status": "verified", "message": "Domain is ready to send!"}

        # Check which records are still missing
        pending_records = [
            r for r in result.data.dns_records
            if not r.verified
        ]

        if attempt < max_attempts - 1:
            # Exponential backoff: 10s, 20s, 40s, 80s...
            wait_time = min(10 * (2 ** attempt), 300)
            time.sleep(wait_time)

    return {
        "status": "pending",
        "message": "DNS records not yet propagated",
        "pending_records": [
            {"type": r.type, "hostname": r.hostname, "purpose": r.purpose}
            for r in pending_records
        ]
    }

# Usage
result = verify_domain_with_polling("dom_xyz789")
if result["status"] == "verified":
    print("✅ Domain verified! Customer can start sending.")
else:
    print(f"⏳ Still waiting on: {result['pending_records']}")

def get_verification_help(pending_records):
    """Generate helpful guidance for pending DNS records."""
    help_messages = []

    for record in pending_records:
        purpose = record.get("purpose", "")

        if purpose == "spf":
            help_messages.append({
                "record": "SPF",
                "issue": "SPF record not found or incorrect",
                "fix": (
                    f"Add a TXT record to {record['hostname']} with value: "
                    f"v=spf1 include:spf.arkhq.io ~all\n"
                    "If you already have an SPF record, add 'include:spf.arkhq.io' "
                    "to your existing record instead of creating a new one."
                )
            })
        elif purpose == "dkim":
            help_messages.append({
                "record": "DKIM",
                "issue": "DKIM record not found",
                "fix": (
                    f"Add a TXT record to {record['hostname']} with the value "
                    "shown in your domain setup page. DKIM values are long — "
                    "make sure you copy the complete value."
                )
            })
        elif purpose == "return_path":
            help_messages.append({
                "record": "Return Path",
                "issue": "Return path CNAME not found",
                "fix": (
                    f"Add a CNAME record for {record['hostname']} pointing to "
                    "rp.arkhq.io"
                )
            })

    return help_messages

from ark import Ark
import ark
import time

client = Ark()

class DomainOnboardingService:
    """Manages domain onboarding for SaaS customer tenants."""

    def start_onboarding(self, tenant_id: str, domain_name: str):
        """Step 1: Create domain and return DNS instructions."""
        try:
            domain = client.domains.create(
                name=domain_name,
                tenant_id=tenant_id
            )
            return {
                "success": True,
                "domain_id": domain.data.id,
                "dns_records": [
                    {
                        "type": r.type,
                        "hostname": r.hostname,
                        "value": r.value,
                        "purpose": r.purpose
                    }
                    for r in domain.data.dns_records
                ]
            }
        except ark.BadRequestError as e:
            return {"success": False, "error": str(e.message)}

    def check_verification(self, domain_id: str):
        """Step 2: Check verification status."""
        result = client.domains.verify(domain_id)
        records = result.data.dns_records

        verified = [r for r in records if r.verified]
        pending = [r for r in records if not r.verified]

        return {
            "status": result.data.verification_status,
            "verified_count": len(verified),
            "total_count": len(records),
            "all_verified": len(pending) == 0,
            "pending": [
                {"purpose": r.purpose, "hostname": r.hostname}
                for r in pending
            ],
            "help": get_verification_help(
                [{"purpose": r.purpose, "hostname": r.hostname} for r in pending]
            ) if pending else []
        }

    def send_test_email(self, tenant_id: str, domain_name: str, to: str):
        """Step 3: Send a test email from the newly verified domain."""
        try:
            email = client.emails.send(
                from_=f"test@{domain_name}",
                to=[to],
                subject=f"Test email from {domain_name}",
                html="<h1>Domain verified!</h1><p>Your domain is set up correctly.</p>",
                text="Domain verified! Your domain is set up correctly.",
                tenant_id=tenant_id
            )
            return {"success": True, "email_id": email.data.id}
        except ark.APIError as e:
            return {"success": False, "error": str(e.message)}

# Usage in your API endpoint
service = DomainOnboardingService()

# Customer enters domain
onboarding = service.start_onboarding("tenant_abc", "mail.customer.com")

# Customer clicks "Verify" after adding DNS records
verification = service.check_verification(onboarding["domain_id"])

if verification["all_verified"]:
    # Send test email
    test = service.send_test_email("tenant_abc", "mail.customer.com", "admin@customer.com")
    print(f"✅ Onboarding complete! Test email: {test['email_id']}")
else:
    print(f"⏳ {verification['verified_count']}/{verification['total_count']} records verified")
    for item in verification["help"]:
        print(f"  → {item['record']}: {item['fix']}")

import asyncio
from datetime import datetime, timedelta

async def background_domain_verification(domain_id: str, tenant_id: str):
    """
    Run in a background job (Celery, etc.).
    Polls verification every 2 minutes for up to 24 hours.
    Notifies the customer when verification succeeds or fails.
    """
    max_duration = timedelta(hours=24)
    start_time = datetime.now()
    poll_interval = 120  # 2 minutes

    while datetime.now() - start_time < max_duration:
        result = client.domains.verify(domain_id)

        if result.data.verification_status == "verified":
            # Notify customer (email, in-app notification, webhook)
            await notify_customer(tenant_id, {
                "type": "domain_verified",
                "domain": result.data.name,
                "message": "Your domain is verified and ready to send!"
            })
            return True

        await asyncio.sleep(poll_interval)

    # Verification timed out after 24 hours
    await notify_customer(tenant_id, {
        "type": "domain_verification_timeout",
        "domain_id": domain_id,
        "message": "Domain verification timed out. Please check your DNS records."
    })
    return False