url_shortening_service/site/links/services.py

"""Link minting, target-URL allowlist, click recording."""

from __future__ import annotations

import hashlib
import hmac
import logging
import secrets
from datetime import datetime
from urllib.parse import urlencode, urlsplit, urlunsplit

from django.conf import settings
from django.db import IntegrityError
from django.db.models import F
from django.http import HttpRequest
from django.utils import timezone
from django.utils.dateparse import parse_datetime

from links.models import Click, ShortLink

logger = logging.getLogger(__name__)

MAX_CODE_ATTEMPTS = 8


class ValidationError(ValueError):
    pass


class CodeCollisionError(RuntimeError):
    pass


def host_allowed(hostname: str, allowed: list[str]) -> bool:
    hostname = hostname.lower().rstrip(".")
    for entry in allowed:
        entry = entry.lower().strip()
        if entry.startswith("*."):
            entry = entry[2:]
        entry = entry.lstrip(".").rstrip(".")
        if not entry:
            continue
        if hostname == entry or hostname.endswith("." + entry):
            return True
    return False


def normalize_destination_host(raw: str) -> str:
    """Strip scheme/path from a domain field. Raise ValidationError if empty."""
    raw = (raw or "").strip()
    if not raw:
        raise ValidationError("invalid url")
    if raw.startswith("//"):
        raise ValidationError("invalid url")
    if "://" not in raw:
        raw = "https://" + raw
    try:
        parts = urlsplit(raw)
    except ValueError as exc:
        raise ValidationError("invalid url") from exc
    hostname = (parts.hostname or "").lower().rstrip(".")
    if not hostname:
        raise ValidationError("invalid url")
    if parts.username or parts.password:
        raise ValidationError("invalid url")
    return hostname


def build_tracked_url(*, domain: str, campaign: str, source: str, metric: str) -> str:
    """Build an allowlisted https URL with UTM query params."""
    hostname = normalize_destination_host(domain)
    query = urlencode(
        {
            "utm_campaign": campaign.strip(),
            "utm_source": source.strip(),
            "utm_medium": metric.strip(),
        }
    )
    return validate_target_url(urlunsplit(("https", hostname, "/", query, "")))


def validate_target_url(raw: str) -> str:
    """Return a canonical https URL or raise ValidationError."""
    if not raw or not isinstance(raw, str):
        raise ValidationError("invalid url")
    raw = raw.strip()
    if raw.startswith("//"):
        raise ValidationError("invalid url")

    try:
        parts = urlsplit(raw)
    except ValueError as exc:
        raise ValidationError("invalid url") from exc

    if parts.scheme.lower() != "https":
        raise ValidationError("invalid url")
    if parts.username or parts.password:
        raise ValidationError("invalid url")

    hostname = (parts.hostname or "").lower().rstrip(".")
    if not hostname:
        raise ValidationError("invalid url")

    allowed = list(getattr(settings, "SHORT_ALLOWED_HOSTS", []) or [])
    if not host_allowed(hostname, allowed):
        raise ValidationError("host not allowlisted")

    netloc = hostname
    if parts.port:
        netloc = f"{hostname}:{parts.port}"
    return urlunsplit(("https", netloc, parts.path, parts.query, parts.fragment))


def generate_code(length: int | None = None) -> str:
    alphabet = settings.CODE_ALPHABET
    size = length if length is not None else settings.SHORT_CODE_LENGTH
    return "".join(secrets.choice(alphabet) for _ in range(size))


def mint_unique_code() -> str:
    for _ in range(MAX_CODE_ATTEMPTS):
        code = generate_code()
        if not ShortLink.objects.filter(code=code).exists():
            return code
    raise CodeCollisionError("could not allocate a unique code")


def parse_expires_at(value) -> datetime | None:
    if value in (None, ""):
        return None
    if not isinstance(value, str):
        raise ValidationError("invalid expires_at")
    parsed = parse_datetime(value)
    if parsed is None:
        raise ValidationError("invalid expires_at")
    if timezone.is_naive(parsed):
        parsed = timezone.make_aware(parsed, timezone.get_current_timezone())
    return parsed


def find_idempotent_link(target_url: str, external_ref: str) -> ShortLink | None:
    if not external_ref:
        return None
    qs = ShortLink.objects.filter(
        target_url=target_url,
        external_ref=external_ref,
        is_active=True,
    )
    now = timezone.now()
    for link in qs:
        if link.expires_at is None or link.expires_at > now:
            return link
    return None


def create_link(
    *,
    target_url: str,
    title: str,
    external_ref: str,
    expires_at: datetime | None,
    token_name: str,
) -> tuple[ShortLink, bool]:
    """Return ``(link, created)``. ``created`` is False on idempotent hit."""
    existing = find_idempotent_link(target_url, external_ref)
    if existing:
        return existing, False

    for _ in range(MAX_CODE_ATTEMPTS):
        try:
            link = ShortLink.objects.create(
                code=generate_code(),
                target_url=target_url,
                title=title,
                external_ref=external_ref,
                expires_at=expires_at,
                created_by_token=token_name,
            )
            return link, True
        except IntegrityError:
            continue
    raise CodeCollisionError("could not allocate a unique code")


def client_ip(request: HttpRequest) -> str:
    forwarded = request.META.get("HTTP_X_FORWARDED_FOR") or ""
    if forwarded:
        return forwarded.split(",")[0].strip()
    return (request.META.get("REMOTE_ADDR") or "").strip()


def hash_ip(ip: str) -> str:
    pepper = getattr(settings, "CLICK_IP_PEPPER", "") or ""
    if not ip or not pepper:
        return ""
    return hmac.new(pepper.encode(), ip.encode(), hashlib.sha256).hexdigest()


def record_click(request: HttpRequest, link: ShortLink) -> None:
    Click.objects.create(
        link=link,
        ip_hash=hash_ip(client_ip(request)),
        user_agent=(request.META.get("HTTP_USER_AGENT") or "")[:512],
        referrer=(request.META.get("HTTP_REFERER") or "")[:1024],
    )
    ShortLink.objects.filter(pk=link.pk).update(click_count=F("click_count") + 1)


def record_click_best_effort(request: HttpRequest, link: ShortLink) -> None:
    try:
        record_click(request, link)
    except Exception:
        logger.exception("click record failed for code=%s", link.code)