deploy: auto-deploy 04:41:22

app/api/server.py

@@ -88,6 +88,7 @@ def create_application() -> FastAPI:
             {"name": "Embeddings", "description": "Text embedding generation using transformer models"},
             {"name": "Verify", "description": "Phone number and identity verification"},
             {"name": "Vector Stores", "description": "Create, manage, and search vector stores for RAG"},
+            {"name": "URL Shortener", "description": "Create and manage short URLs with analytics"},
         ],
         lifespan=lifespan,
     )

app/api/v1/router.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from fastapi import APIRouter
 
-from app.api.v1 import auth, batch, chat, code_executor, convert, database, embeddings, reconcile, scraper, semantic_router, sql_validator, system, token_counter, token_generator, vector_stores, web_search, webhook_socket
+from app.api.v1 import auth, batch, chat, code_executor, convert, database, embeddings, reconcile, scraper, semantic_router, sql_validator, system, token_counter, token_generator, url_shortener, vector_stores, web_search, webhook_socket
 from app.api.verify import router as verify_router
 
 api_v1_router = APIRouter()
@@ -24,3 +24,4 @@ api_v1_router.include_router(token_generator.router, tags=["Token Generator"])
 api_v1_router.include_router(chat.router, tags=["Chat"])
 api_v1_router.include_router(vector_stores.router, tags=["Vector Stores"])
 api_v1_router.include_router(webhook_socket.router, tags=["Webhook / Socket"])
+api_v1_router.include_router(url_shortener.router, tags=["URL Shortener"])

app/api/v1/url_shortener.py

@@ -0,0 +1,663 @@
+from __future__ import annotations
+
+import threading
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, Depends, Header, HTTPException, Path, Query
+from fastapi.responses import PlainTextResponse
+from pydantic import BaseModel, Field
+
+from app.config import get_settings
+from app.services.url_shortener_service import (
+    URLShortenerService,
+    AliasTakenError,
+    AuthenticationError,
+    AuthorizationError,
+    LinkExpiredError,
+    LinkNotFoundError,
+    PlanLimitExceededError,
+    URLShortenerError,
+    ValidationError,
+)
+
+_settings = get_settings()
+router = APIRouter()
+
+_service_lock = threading.Lock()
+_shared_service: Optional[URLShortenerService] = None
+
+
+def _get_service() -> URLShortenerService:
+    global _shared_service
+    if _shared_service is None:
+        with _service_lock:
+            if _shared_service is None:
+                _shared_service = URLShortenerService()
+    return _shared_service
+
+
+def _app_id() -> str:
+    return _settings.application_id or ""
+
+
+# ---------------------------------------------------------------------------
+# Request / Response models
+# ---------------------------------------------------------------------------
+
+class CreateOwnerRequest(BaseModel):
+    name: str = Field(..., min_length=1, max_length=200)
+    plan: Optional[str] = Field("free", description="free, pro, or enterprise")
+
+
+class CreateOwnerResponse(BaseModel):
+    success: bool = True
+    owner_id: str
+    api_key: str
+    plan: str
+    application_id: str = ""
+
+
+class ShortenRequest(BaseModel):
+    long_url: str = Field(..., description="Destination URL (http/https only)")
+    custom_alias: Optional[str] = Field(None, description="Custom short code (3-32 chars, letters/digits/-/_)")
+    expires_in_days: Optional[int] = Field(None, ge=1)
+    max_clicks: Optional[int] = Field(None, ge=1)
+    tags: Optional[List[str]] = Field(None, description="Tags (max 20)")
+    note: Optional[str] = Field(None, description="Max 2000 chars")
+    password: Optional[str] = Field(None, description="Password-protect the link")
+    utm_source: Optional[str] = Field(None)
+    utm_medium: Optional[str] = Field(None)
+    utm_campaign: Optional[str] = Field(None)
+    campaign_id: Optional[str] = Field(None, description="Campaign to group this link under")
+    custom_domain: Optional[str] = Field(None, description="Branded domain override")
+    fallback_url: Optional[str] = Field(None, description="Redirect here when link is expired/deactivated")
+    webhook_url: Optional[str] = Field(None, description="URL to POST click events to (Enterprise)")
+    geo_targeting: Optional[Dict[str, Any]] = Field(None, description='{"countries": {...}, "devices": {...}}')
+
+
+class ShortLinkResponse(BaseModel):
+    success: bool = True
+    short_code: str
+    long_url: str
+    owner_id: str
+    created_at: str
+    expires_at: Optional[str] = None
+    max_clicks: Optional[int] = None
+    click_count: int = 0
+    is_active: bool = True
+    tags: List[str] = []
+    note: Optional[str] = None
+    has_password: bool = False
+    short_url: Optional[str] = None
+    qr_image_url: Optional[str] = None
+    campaign_id: Optional[str] = None
+    custom_domain: Optional[str] = None
+    fallback_url: Optional[str] = None
+    webhook_url: Optional[str] = None
+    application_id: str = ""
+
+
+class UpdateLinkRequest(BaseModel):
+    long_url: Optional[str] = None
+    note: Optional[str] = None
+    tags: Optional[List[str]] = None
+    max_clicks: Optional[int] = None
+    password: Optional[str] = None
+    webhook_url: Optional[str] = None
+    fallback_url: Optional[str] = None
+    campaign_id: Optional[str] = None
+    custom_domain: Optional[str] = None
+    geo_targeting: Optional[Dict[str, Any]] = None
+
+
+class ListLinksResponse(BaseModel):
+    success: bool = True
+    links: List[ShortLinkResponse]
+    application_id: str = ""
+
+
+class StatsResponse(BaseModel):
+    success: bool = True
+    short_code: str
+    long_url: str
+    click_count: int
+    created_at: str
+    last_accessed_at: Optional[str] = None
+    is_active: bool
+    expires_at: Optional[str] = None
+    max_clicks: Optional[int] = None
+    has_password: bool = False
+    utm: Optional[Dict[str, Any]] = None
+    analytics: Optional[Dict[str, Any]] = None
+    campaign_id: Optional[str] = None
+    custom_domain: Optional[str] = None
+    fallback_url: Optional[str] = None
+    webhook_url: Optional[str] = None
+    application_id: str = ""
+
+
+class ClickDetail(BaseModel):
+    id: int
+    referrer: Optional[str] = None
+    user_agent: Optional[str] = None
+    ip_address: Optional[str] = None
+    browser: Optional[str] = None
+    device: Optional[str] = None
+    os: Optional[str] = None
+    clicked_at: str
+
+
+class ClickDetailsResponse(BaseModel):
+    success: bool = True
+    clicks: List[ClickDetail]
+    application_id: str = ""
+
+
+class QRCodeResponse(BaseModel):
+    success: bool = True
+    short_code: str
+    short_url: str
+    qr_image_url: str
+    application_id: str = ""
+
+
+class ErrorResponse(BaseModel):
+    success: bool = False
+    error: str
+
+
+class BulkShortenRequest(BaseModel):
+    items: List[Dict[str, Any]] = Field(..., description="Array of shorten requests (max 100)")
+
+
+class BulkShortenResponse(BaseModel):
+    success: bool = True
+    results: List[Dict[str, Any]]
+    application_id: str = ""
+
+
+class OwnerSummaryResponse(BaseModel):
+    success: bool = True
+    owner_id: str
+    name: str
+    plan: str
+    total_links: int
+    active_links: int
+    total_clicks: int
+    created_at: str
+    application_id: str = ""
+
+
+class CampaignCreateRequest(BaseModel):
+    name: str = Field(..., min_length=1, max_length=200)
+    description: Optional[str] = Field(None, max_length=2000)
+
+
+class CampaignResponse(BaseModel):
+    success: bool = True
+    campaign_id: str
+    name: str
+    description: Optional[str] = None
+    created_at: Optional[str] = None
+    is_active: Optional[bool] = None
+    link_count: Optional[int] = 0
+    total_clicks: Optional[int] = 0
+    analytics: Optional[Dict[str, Any]] = None
+    application_id: str = ""
+
+
+class CampaignListResponse(BaseModel):
+    success: bool = True
+    campaigns: List[CampaignResponse]
+    application_id: str = ""
+
+
+class HealthCheckResponse(BaseModel):
+    success: bool = True
+    short_code: str
+    long_url: str
+    reachable: bool
+    status_code: Optional[int] = None
+    error: Optional[str] = None
+    application_id: str = ""
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _auth_owner(x_api_key: str = Header(..., alias="X-API-Key")) -> str:
+    service = _get_service()
+    try:
+        return service.authenticate(x_api_key)
+    except AuthenticationError:
+        raise HTTPException(status_code=401, detail={"success": False, "error": "Invalid API key"})
+
+
+def _build_link_response(link, service) -> ShortLinkResponse:
+    return ShortLinkResponse(
+        short_code=link.short_code,
+        long_url=link.long_url,
+        owner_id=link.owner_id,
+        created_at=link.created_at.isoformat(),
+        expires_at=link.expires_at.isoformat() if link.expires_at else None,
+        max_clicks=link.max_clicks,
+        click_count=link.click_count,
+        is_active=link.is_active,
+        tags=link.tags,
+        note=link.note,
+        has_password=bool(link.password_hash) if hasattr(link, 'password_hash') else False,
+        campaign_id=link.campaign_id,
+        custom_domain=link.custom_domain,
+        fallback_url=link.fallback_url,
+        webhook_url=link.webhook_url,
+        application_id=_app_id(),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Owner / Plan
+# ---------------------------------------------------------------------------
+
+@router.post(
+    "/url-shortener/owners",
+    response_model=CreateOwnerResponse,
+    summary="Create a new owner (tenant) and issue API key",
+)
+def create_owner(body: CreateOwnerRequest):
+    service = _get_service()
+    result = service.create_owner(body.name, body.plan or "free")
+    return CreateOwnerResponse(
+        owner_id=result["owner_id"],
+        api_key=result["api_key"],
+        plan=body.plan or "free",
+        application_id=_app_id(),
+    )
+
+
+@router.patch(
+    "/url-shortener/plan",
+    summary="Update owner's subscription plan",
+)
+def update_plan(plan: str = Query(..., description="free, pro, or enterprise"),
+                owner_id: str = Depends(_auth_owner)):
+    service = _get_service()
+    try:
+        return {"success": True, "application_id": _app_id(), **service.update_owner_plan(owner_id, plan)}
+    except ValidationError as e:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(e)})
+
+
+@router.get(
+    "/url-shortener/summary",
+    response_model=OwnerSummaryResponse,
+    summary="Get owner dashboard summary",
+)
+def owner_summary(owner_id: str = Depends(_auth_owner)):
+    service = _get_service()
+    data = service.owner_summary(owner_id)
+    if not data:
+        raise HTTPException(status_code=404, detail={"success": False, "error": "Owner not found"})
+    return OwnerSummaryResponse(application_id=_app_id(), **data)
+
+
+# ---------------------------------------------------------------------------
+# Short links CRUD
+# ---------------------------------------------------------------------------
+
+@router.post(
+    "/url-shortener/shorten",
+    response_model=ShortLinkResponse,
+    summary="Create a short link",
+)
+def shorten_url(body: ShortenRequest, owner_id: str = Depends(_auth_owner)):
+    service = _get_service()
+    try:
+        link = service.shorten(
+            long_url=body.long_url,
+            owner_id=owner_id,
+            custom_alias=body.custom_alias,
+            expires_in_days=body.expires_in_days,
+            max_clicks=body.max_clicks,
+            tags=body.tags,
+            note=body.note,
+            password=body.password,
+            utm_source=body.utm_source,
+            utm_medium=body.utm_medium,
+            utm_campaign=body.utm_campaign,
+            campaign_id=body.campaign_id,
+            custom_domain=body.custom_domain,
+            fallback_url=body.fallback_url,
+            webhook_url=body.webhook_url,
+            geo_targeting=body.geo_targeting,
+        )
+        from app.services.url_shortener_service import SRV_BASE_URL
+        resp = _build_link_response(link, service)
+        short_url = f"{SRV_BASE_URL}/url-shortener/{link.short_code}"
+        if link.custom_domain:
+            short_url = f"https://{link.custom_domain}/{link.short_code}"
+        resp.short_url = short_url
+        resp.qr_image_url = f"https://api.qrserver.com/v1/create-qr-code/?size=300x300&data={short_url}"
+        return resp
+    except ValidationError as e:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(e)})
+    except AuthenticationError as e:
+        raise HTTPException(status_code=401, detail={"success": False, "error": str(e)})
+    except AliasTakenError as e:
+        raise HTTPException(status_code=409, detail={"success": False, "error": str(e)})
+    except PlanLimitExceededError as e:
+        raise HTTPException(status_code=402, detail={"success": False, "error": str(e), "code": "plan_limit"})
+    except URLShortenerError as e:
+        raise HTTPException(status_code=500, detail={"success": False, "error": str(e)})
+
+
+@router.post(
+    "/url-shortener/bulk",
+    response_model=BulkShortenResponse,
+    summary="Bulk create short links",
+)
+def bulk_shorten(body: BulkShortenRequest, owner_id: str = Depends(_auth_owner)):
+    if len(body.items) > 100:
+        raise HTTPException(status_code=400, detail={"success": False, "error": "Bulk limit is 100 items per request"})
+    service = _get_service()
+    results = service.bulk_shorten(owner_id, body.items)
+    return BulkShortenResponse(results=results, application_id=_app_id())
+
+
+@router.get(
+    "/url-shortener/links",
+    response_model=ListLinksResponse,
+    summary="List all links for the authenticated owner",
+)
+def list_links(owner_id: str = Depends(_auth_owner)):
+    service = _get_service()
+    links = service.list_links(owner_id)
+    items = [_build_link_response(link, service) for link in links]
+    return ListLinksResponse(links=items, application_id=_app_id())
+
+
+@router.get(
+    "/url-shortener/export",
+    summary="Export all links as CSV",
+)
+def export_links_csv(owner_id: str = Depends(_auth_owner)):
+    service = _get_service()
+    csv_data = service.export_links(owner_id)
+    return PlainTextResponse(csv_data, media_type="text/csv",
+                             headers={"Content-Disposition": "attachment; filename=links_export.csv"})
+
+
+@router.post(
+    "/url-shortener/campaigns",
+    response_model=CampaignResponse,
+    summary="Create a campaign to group links",
+)
+def create_campaign(body: CampaignCreateRequest, owner_id: str = Depends(_auth_owner)):
+    service = _get_service()
+    try:
+        result = service.create_campaign(owner_id, body.name, body.description)
+        return CampaignResponse(
+            campaign_id=result["campaign_id"],
+            name=result["name"],
+            description=result.get("description"),
+            application_id=_app_id(),
+        )
+    except PlanLimitExceededError as e:
+        raise HTTPException(status_code=402, detail={"success": False, "error": str(e), "code": "plan_limit"})
+    except ValidationError as e:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(e)})
+
+
+@router.get(
+    "/url-shortener/campaigns",
+    response_model=CampaignListResponse,
+    summary="List all campaigns for the authenticated owner",
+)
+def list_campaigns(owner_id: str = Depends(_auth_owner)):
+    service = _get_service()
+    campaigns = service.list_campaigns(owner_id)
+    items = [CampaignResponse(**c, application_id=_app_id()) for c in campaigns]
+    return CampaignListResponse(campaigns=items, application_id=_app_id())
+
+
+@router.get(
+    "/url-shortener/campaigns/{campaign_id}",
+    response_model=CampaignResponse,
+    summary="Get campaign details with analytics",
+)
+def get_campaign(
+    campaign_id: str = Path(...),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        data = service.get_campaign(campaign_id, owner_id)
+        return CampaignResponse(**data, application_id=_app_id())
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+@router.get(
+    "/url-shortener/campaigns/{campaign_id}/links",
+    response_model=ListLinksResponse,
+    summary="List all links in a campaign",
+)
+def list_campaign_links(
+    campaign_id: str = Path(...),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        links = service.list_campaign_links(campaign_id, owner_id)
+        items = [_build_link_response(link, service) for link in links]
+        return ListLinksResponse(links=items, application_id=_app_id())
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+@router.delete(
+    "/url-shortener/campaigns/{campaign_id}",
+    summary="Deactivate a campaign",
+)
+def deactivate_campaign(
+    campaign_id: str = Path(...),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        service.deactivate_campaign(campaign_id, owner_id)
+        return {"success": True, "message": f"Campaign '{campaign_id}' deactivated", "application_id": _app_id()}
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+# ---------------------------------------------------------------------------
+# Short link path-parameter routes (must come after static routes)
+# ---------------------------------------------------------------------------
+
+@router.get(
+    "/url-shortener/{short_code}",
+    summary="Resolve a short code (redirect to long URL)",
+)
+def resolve_short_code(
+    short_code: str = Path(...),
+    x_link_password: Optional[str] = Header(None, alias="X-Link-Password"),
+    referer: Optional[str] = Header(None, alias="Referer"),
+    user_agent: Optional[str] = Header(None, alias="User-Agent"),
+    x_forwarded_for: Optional[str] = Header(None, alias="X-Forwarded-For"),
+):
+    service = _get_service()
+    from fastapi.responses import RedirectResponse
+    ip = x_forwarded_for.split(",")[0].strip() if x_forwarded_for else None
+    try:
+        if x_link_password:
+            long_url = service.resolve_with_password(short_code, x_link_password, referer, user_agent, ip)
+        else:
+            long_url = service.resolve(short_code, referer, user_agent, ip)
+        return RedirectResponse(url=long_url, status_code=302)
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except LinkExpiredError as e:
+        raise HTTPException(status_code=410, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+@router.patch(
+    "/url-shortener/{short_code}",
+    response_model=ShortLinkResponse,
+    summary="Update a short link (change URL, tags, webhook, etc.)",
+)
+def update_short_link(
+    body: UpdateLinkRequest,
+    short_code: str = Path(...),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        updates = {k: v for k, v in body.model_dump().items() if v is not None}
+        if not updates:
+            raise HTTPException(status_code=400, detail={"success": False, "error": "No fields to update"})
+        link = service.update_link(short_code, owner_id, **updates)
+        return _build_link_response(link, service)
+    except (ValidationError, AuthenticationError) as e:
+        raise HTTPException(status_code=400, detail={"success": False, "error": str(e)})
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+@router.delete(
+    "/url-shortener/{short_code}",
+    summary="Deactivate a short link",
+)
+def deactivate_link(
+    short_code: str = Path(...),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        service.deactivate_link(short_code, owner_id)
+        return {"success": True, "message": f"Link '{short_code}' deactivated", "application_id": _app_id()}
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+@router.get(
+    "/url-shortener/{short_code}/stats",
+    response_model=StatsResponse,
+    summary="Get click statistics for a link (browsers, devices, referrers, etc.)",
+)
+def get_link_stats(
+    short_code: str = Path(...),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        stats = service.get_link_stats(short_code, owner_id)
+        return StatsResponse(
+            short_code=stats["short_code"],
+            long_url=stats["long_url"],
+            click_count=stats["click_count"],
+            created_at=stats["created_at"],
+            last_accessed_at=stats["last_accessed_at"],
+            is_active=stats["is_active"],
+            expires_at=stats["expires_at"],
+            max_clicks=stats["max_clicks"],
+            has_password=stats["has_password"],
+            utm=stats["utm"],
+            analytics=stats["analytics"],
+            campaign_id=stats.get("campaign_id"),
+            custom_domain=stats.get("custom_domain"),
+            fallback_url=stats.get("fallback_url"),
+            webhook_url=stats.get("webhook_url"),
+            application_id=_app_id(),
+        )
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+@router.get(
+    "/url-shortener/{short_code}/clicks",
+    response_model=ClickDetailsResponse,
+    summary="Get detailed click log for a link",
+)
+def get_click_details(
+    short_code: str = Path(...),
+    limit: int = Query(100, ge=1, le=1000),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        clicks = service.get_click_details(short_code, owner_id, limit)
+        return ClickDetailsResponse(clicks=[ClickDetail(**c) for c in clicks], application_id=_app_id())
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+@router.get(
+    "/url-shortener/{short_code}/qr",
+    response_model=QRCodeResponse,
+    summary="Get QR code for a short link",
+)
+def get_qr_code(
+    short_code: str = Path(...),
+):
+    service = _get_service()
+    try:
+        data = service.get_qr_code(short_code)
+        return QRCodeResponse(**data, application_id=_app_id())
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+
+
+@router.get(
+    "/url-shortener/{short_code}/health",
+    response_model=HealthCheckResponse,
+    summary="Check if the destination URL is reachable",
+)
+def link_health(
+    short_code: str = Path(...),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        result = service.check_link_health(short_code, owner_id)
+        return HealthCheckResponse(**result, application_id=_app_id())
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})
+
+
+@router.get(
+    "/url-shortener/{short_code}/clicks/export",
+    summary="Export click data as CSV",
+)
+def export_clicks_csv(
+    short_code: str = Path(...),
+    owner_id: str = Depends(_auth_owner),
+):
+    service = _get_service()
+    try:
+        csv_data = service.export_clicks(short_code, owner_id)
+        return PlainTextResponse(csv_data, media_type="text/csv",
+                                 headers={"Content-Disposition": f"attachment; filename=clicks_{short_code}.csv"})
+    except LinkNotFoundError as e:
+        raise HTTPException(status_code=404, detail={"success": False, "error": str(e)})
+    except AuthorizationError as e:
+        raise HTTPException(status_code=403, detail={"success": False, "error": str(e)})

app/services/url_shortener_service.py

@@ -0,0 +1,1198 @@
+from __future__ import annotations
+
+import asyncio
+import csv
+import hashlib
+import hmac
+import io
+import ipaddress
+import json
+import os
+import secrets
+import socket
+import sqlite3
+import string
+import threading
+import unicodedata
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional
+from urllib.parse import urlparse
+
+from app.config import get_settings
+from app.core.logger import get_logger
+
+logger = get_logger(__name__)
+
+_Settings = get_settings()
+
+_SHORTENER_SECRET = os.environ.get("URL_SHORTENER_SECRET", "")
+if not _SHORTENER_SECRET or _SHORTENER_SECRET == "dev-secret-change-me":
+    logger.warning(
+        "URL_SHORTENER_SECRET is not set or is the default value. "
+        "Set it to a strong random secret before deploying."
+    )
+
+SRV_BASE_URL = os.environ.get("URL_SHORTENER_BASE", "http://localhost:7860/api/v1")
+
+PLAN_LIMITS: Dict[str, Dict[str, Any]] = {
+    "free": {"max_links": 50, "max_clicks_per_link": 1000, "custom_alias": False, "analytics_days": 7, "tags": False, "max_api_keys": 1, "campaigns": False, "webhooks": False, "custom_domain": False, "geo_targeting": False},
+    "pro": {"max_links": 5000, "max_clicks_per_link": 100000, "custom_alias": True, "analytics_days": 365, "tags": True, "max_api_keys": 5, "campaigns": True, "webhooks": False, "custom_domain": False, "geo_targeting": False},
+    "enterprise": {"max_links": 999999, "max_clicks_per_link": 99999999, "custom_alias": True, "analytics_days": 9999, "tags": True, "max_api_keys": 50, "campaigns": True, "webhooks": True, "custom_domain": True, "geo_targeting": True},
+}
+
+
+@dataclass
+class ShortLink:
+    short_code: str
+    long_url: str
+    owner_id: str
+    created_at: datetime
+    expires_at: Optional[datetime]
+    max_clicks: Optional[int]
+    click_count: int
+    is_active: bool
+    tags: List[str]
+    note: Optional[str]
+    password_hash: Optional[str] = None
+    utm_source: Optional[str] = None
+    utm_medium: Optional[str] = None
+    utm_campaign: Optional[str] = None
+    campaign_id: Optional[str] = None
+    custom_domain: Optional[str] = None
+    fallback_url: Optional[str] = None
+    webhook_url: Optional[str] = None
+    geo_targeting: Optional[Dict[str, Any]] = None
+
+
+class URLShortenerError(Exception):
+    pass
+
+class ValidationError(URLShortenerError):
+    pass
+
+class AliasTakenError(URLShortenerError):
+    pass
+
+class LinkNotFoundError(URLShortenerError):
+    pass
+
+class LinkExpiredError(URLShortenerError):
+    pass
+
+class AuthenticationError(URLShortenerError):
+    pass
+
+class AuthorizationError(URLShortenerError):
+    pass
+
+class PlanLimitExceededError(URLShortenerError):
+    pass
+
+
+_ALPHABET = string.digits + string.ascii_lowercase + string.ascii_uppercase
+
+_BLOCKED_DNS_WILDCARDS = (
+    "nip.io", "sslip.io", "xip.io", "localtest.me",
+    "lvh.me", "1u.ms", "traefik.me",
+)
+
+
+def _is_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    if ip.version == 6:
+        mapped = ip.ipv4_mapped
+        if mapped is not None:
+            if _is_blocked_ip(mapped):
+                return True
+        ip_int = int(ip)
+        if ip_int >> 32 == 0:
+            ipv4_part = ipaddress.IPv4Address(ip_int & 0xFFFFFFFF)
+            if _is_blocked_ip(ipv4_part):
+                return True
+        return False
+    addr_int = int(ip)
+    if addr_int >> 24 == 0:
+        return True
+    return (
+        ip.is_private or ip.is_loopback or ip.is_link_local
+        or ip.is_reserved or ip.is_multicast
+    )
+
+
+def _check_ip_string(hostname: str) -> bool:
+    try:
+        return _is_blocked_ip(ipaddress.ip_address(hostname))
+    except ValueError:
+        pass
+    if hostname.isdigit():
+        try:
+            return _is_blocked_ip(ipaddress.ip_address(int(hostname)))
+        except (ValueError, OverflowError):
+            pass
+    if hostname.lower().startswith("0x"):
+        try:
+            return _is_blocked_ip(ipaddress.ip_address(int(hostname, 16)))
+        except (ValueError, OverflowError):
+            pass
+    parts = hostname.split(".")
+    if all(p.lower().startswith("0x") for p in parts) and len(parts) in (2, 3, 4):
+        try:
+            numeric = sum(int(p, 16) << (8 * (len(parts) - 1 - i)) for i, p in enumerate(parts))
+            return _is_blocked_ip(ipaddress.ip_address(numeric))
+        except (ValueError, OverflowError):
+            pass
+    if 1 <= len(parts) <= 3 and all(p.isdigit() for p in parts):
+        expanded = parts + ["0"] * (4 - len(parts))
+        try:
+            return _is_blocked_ip(ipaddress.ip_address(".".join(expanded)))
+        except ValueError:
+            pass
+    if len(parts) == 4:
+        for p in parts:
+            if len(p) > 1 and p[0] == "0" and p.isdigit():
+                return True
+    return False
+
+
+def _normalize_hostname(hostname: str) -> str:
+    return unicodedata.normalize("NFKC", hostname).lower()
+
+
+def _has_dns_wildcard_suffix(hostname: str, suffixes: tuple) -> bool:
+    hostname = _normalize_hostname(hostname)
+    for suffix in suffixes:
+        if hostname == suffix or hostname.endswith("." + suffix):
+            return True
+    return False
+
+
+def validate_url(url: str, max_length: int = 2048) -> None:
+    if not url or not isinstance(url, str):
+        raise ValidationError("URL must be a non-empty string")
+    stripped = url.strip()
+    if stripped != url:
+        raise ValidationError("URL must not contain leading or trailing whitespace")
+    if len(url) > max_length:
+        raise ValidationError(f"URL exceeds max length of {max_length}")
+    if "@" in url:
+        raise ValidationError("URL must not contain userinfo ('@')")
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise ValidationError("Only http and https schemes are allowed")
+    hostname = _normalize_hostname(parsed.hostname or "")
+    if not hostname:
+        raise ValidationError("URL must include a valid host")
+    if hostname == "localhost":
+        raise ValidationError("URLs pointing at localhost are not allowed")
+    if _has_dns_wildcard_suffix(hostname, _BLOCKED_DNS_WILDCARDS):
+        raise ValidationError(f"URL uses a blocked DNS wildcard service ({hostname})")
+    if _check_ip_string(hostname):
+        raise ValidationError("URLs pointing at private/internal addresses are not allowed")
+    try:
+        addrinfo = socket.getaddrinfo(hostname, None)
+    except (socket.gaierror, OSError):
+        addrinfo = ()
+    for family, _, _, _, sockaddr in addrinfo:
+        ip_str = sockaddr[0]
+        try:
+            ip = ipaddress.ip_address(ip_str)
+            if _is_blocked_ip(ip):
+                raise ValidationError(f"URL resolves to a private/internal address ({ip_str})")
+        except ValueError:
+            continue
+
+
+def validate_alias(alias: str) -> None:
+    if not (3 <= len(alias) <= 32):
+        raise ValidationError("Custom alias must be 3-32 characters")
+    allowed = set(string.ascii_letters + string.digits + "-_")
+    if not set(alias) <= allowed:
+        raise ValidationError("Custom alias may only contain letters, digits, '-' and '_'")
+
+
+def generate_short_code(length: int) -> str:
+    if length < 1:
+        raise ValueError("Short code length must be at least 1")
+    return "".join(secrets.choice(_ALPHABET) for _ in range(length))
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _parse_iso(value: Optional[str]) -> Optional[datetime]:
+    return datetime.fromisoformat(value) if value else None
+
+
+def _hash_key(api_key: str) -> str:
+    secret = os.environ.get("URL_SHORTENER_SECRET", "dev-secret-change-me").encode()
+    return hmac.new(secret, api_key.encode(), hashlib.sha256).hexdigest()
+
+
+def _hash_password(password: str) -> str:
+    return hashlib.sha256(password.encode()).hexdigest()
+
+
+def _parse_user_agent(ua: str) -> Dict[str, str]:
+    ua_lower = ua.lower()
+    result: Dict[str, str] = {"browser": "Unknown", "device": "Desktop", "os": "Unknown"}
+    if "mobile" in ua_lower or "android" in ua_lower or "iphone" in ua_lower:
+        result["device"] = "Mobile"
+    if "tablet" in ua_lower or "ipad" in ua_lower:
+        result["device"] = "Tablet"
+    if "chrome" in ua_lower and "edge" not in ua_lower:
+        result["browser"] = "Chrome"
+    elif "firefox" in ua_lower:
+        result["browser"] = "Firefox"
+    elif "safari" in ua_lower and "chrome" not in ua_lower:
+        result["browser"] = "Safari"
+    elif "edge" in ua_lower:
+        result["browser"] = "Edge"
+    elif "msie" in ua_lower or "trident" in ua_lower:
+        result["browser"] = "Internet Explorer"
+    if "windows" in ua_lower:
+        result["os"] = "Windows"
+    elif "mac" in ua_lower:
+        result["os"] = "macOS"
+    elif "linux" in ua_lower:
+        result["os"] = "Linux"
+    elif "android" in ua_lower:
+        result["os"] = "Android"
+    elif "ios" in ua_lower or "iphone" in ua_lower:
+        result["os"] = "iOS"
+    return result
+
+
+def _build_qr_data_url(short_code: str) -> str:
+    short_url = f"{SRV_BASE_URL}/url-shortener/{short_code}"
+    return f"https://api.qrserver.com/v1/create-qr-code/?size=300x300&data={short_url}"
+
+
+def _apply_geo_targeting(rules: Dict[str, Any], default_url: str,
+                          ip_address: Optional[str] = None,
+                          user_agent: Optional[str] = None) -> str:
+    countries = rules.get("countries", {})
+    devices = rules.get("devices", {})
+    if not countries and not devices:
+        return default_url
+    ua_lower = (user_agent or "").lower()
+    is_mobile = "mobile" in ua_lower or "android" in ua_lower or "iphone" in ua_lower
+    is_tablet = "tablet" in ua_lower or "ipad" in ua_lower
+    # Device rule takes precedence
+    if is_tablet and "tablet" in devices:
+        return devices["tablet"]
+    if is_mobile and "mobile" in devices:
+        return devices["mobile"]
+    if not (is_mobile or is_tablet) and "desktop" in devices:
+        return devices["desktop"]
+    # Country-specific override (requires IP geolocation service)
+    # For now, limited to simplistic check
+    return default_url
+
+
+def _fire_click_webhook(webhook_url: str, short_code: str, long_url: str,
+                         referrer: Optional[str], user_agent: Optional[str],
+                         ip_address: Optional[str]) -> None:
+    payload = {
+        "event": "link_clicked",
+        "short_code": short_code,
+        "long_url": long_url,
+        "referrer": referrer,
+        "user_agent": user_agent,
+        "ip_address": ip_address,
+        "timestamp": _now_iso(),
+    }
+    try:
+        import httpx
+        with httpx.Client(timeout=5.0) as client:
+            client.post(webhook_url, json=payload)
+    except ImportError:
+        import urllib.request
+        import json as _json
+        data = _json.dumps(payload).encode()
+        req = urllib.request.Request(webhook_url, data=data,
+                                     headers={"Content-Type": "application/json"},
+                                     method="POST")
+        urllib.request.urlopen(req, timeout=5)
+
+
+class Storage:
+    def __init__(self, db_path: str):
+        self._db_path = db_path
+        self._lock = threading.RLock()
+        self._local = threading.local()
+        self._init_schema()
+
+    @property
+    def _conn(self) -> sqlite3.Connection:
+        if not hasattr(self._local, "conn"):
+            conn = sqlite3.connect(self._db_path, check_same_thread=False)
+            conn.row_factory = sqlite3.Row
+            conn.execute("PRAGMA foreign_keys = ON")
+            conn.execute("PRAGMA journal_mode = WAL")
+            self._local.conn = conn
+        return self._local.conn
+
+    @contextmanager
+    def writer_lock(self):
+        with self._lock:
+            yield
+
+    def _init_schema(self) -> None:
+        with self._lock, self._conn as conn:
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS url_shortener_owners (
+                    owner_id      TEXT PRIMARY KEY,
+                    name          TEXT NOT NULL,
+                    api_key_hash  TEXT NOT NULL UNIQUE,
+                    plan          TEXT NOT NULL DEFAULT 'free',
+                    created_at    TEXT NOT NULL
+                )
+            """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS url_shortener_links (
+                    short_code       TEXT PRIMARY KEY,
+                    long_url         TEXT NOT NULL,
+                    owner_id         TEXT NOT NULL,
+                    created_at       TEXT NOT NULL,
+                    expires_at       TEXT,
+                    max_clicks       INTEGER,
+                    click_count      INTEGER NOT NULL DEFAULT 0,
+                    last_accessed_at TEXT,
+                    is_active        INTEGER NOT NULL DEFAULT 1,
+                    tags             TEXT,
+                    note             TEXT,
+                    password_hash    TEXT,
+                    utm_source       TEXT,
+                    utm_medium       TEXT,
+                    utm_campaign     TEXT,
+                    campaign_id      TEXT,
+                    custom_domain    TEXT,
+                    fallback_url     TEXT,
+                    webhook_url      TEXT,
+                    geo_targeting    TEXT,
+                    FOREIGN KEY (owner_id) REFERENCES url_shortener_owners(owner_id)
+                )
+            """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS url_shortener_clicks (
+                    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+                    short_code  TEXT NOT NULL,
+                    referrer    TEXT,
+                    user_agent  TEXT,
+                    ip_address  TEXT,
+                    country     TEXT,
+                    browser     TEXT,
+                    device      TEXT,
+                    os          TEXT,
+                    clicked_at  TEXT NOT NULL,
+                    FOREIGN KEY (short_code) REFERENCES url_shortener_links(short_code)
+                )
+            """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS url_shortener_audit_log (
+                    id         INTEGER PRIMARY KEY AUTOINCREMENT,
+                    timestamp  TEXT NOT NULL,
+                    owner_id   TEXT,
+                    action     TEXT NOT NULL,
+                    short_code TEXT,
+                    detail     TEXT
+                )
+            """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS url_shortener_campaigns (
+                    campaign_id  TEXT PRIMARY KEY,
+                    owner_id     TEXT NOT NULL,
+                    name         TEXT NOT NULL,
+                    description  TEXT,
+                    created_at   TEXT NOT NULL,
+                    is_active    INTEGER NOT NULL DEFAULT 1,
+                    FOREIGN KEY (owner_id) REFERENCES url_shortener_owners(owner_id)
+                )
+            """)
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_us_links_owner ON url_shortener_links(owner_id)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_us_clicks_code ON url_shortener_clicks(short_code)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_us_clicks_at ON url_shortener_clicks(clicked_at)")
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_us_links_campaign ON url_shortener_links(campaign_id)")
+            self._migrate_schema(conn)
+
+    def _migrate_schema(self, conn: sqlite3.Connection) -> None:
+        existing = {r["name"] for r in conn.execute("PRAGMA table_info(url_shortener_links)").fetchall()}
+        migrations = {
+            "campaign_id": "ALTER TABLE url_shortener_links ADD COLUMN campaign_id TEXT",
+            "custom_domain": "ALTER TABLE url_shortener_links ADD COLUMN custom_domain TEXT",
+            "fallback_url": "ALTER TABLE url_shortener_links ADD COLUMN fallback_url TEXT",
+            "webhook_url": "ALTER TABLE url_shortener_links ADD COLUMN webhook_url TEXT",
+            "geo_targeting": "ALTER TABLE url_shortener_links ADD COLUMN geo_targeting TEXT",
+        }
+        for col, ddl in migrations.items():
+            if col not in existing:
+                conn.execute(ddl)
+
+    def create_owner(self, owner_id: str, name: str, api_key_hash: str, plan: str = "free") -> None:
+        with self._lock, self._conn as conn:
+            conn.execute(
+                "INSERT INTO url_shortener_owners (owner_id, name, api_key_hash, plan, created_at) "
+                "VALUES (?, ?, ?, ?, ?)",
+                (owner_id, name, api_key_hash, plan, _now_iso()),
+            )
+
+    def get_owner_by_key_hash(self, api_key_hash: str) -> Optional[sqlite3.Row]:
+        return self._conn.execute(
+            "SELECT * FROM url_shortener_owners WHERE api_key_hash = ?", (api_key_hash,)
+        ).fetchone()
+
+    def get_owner(self, owner_id: str) -> Optional[sqlite3.Row]:
+        return self._conn.execute(
+            "SELECT * FROM url_shortener_owners WHERE owner_id = ?", (owner_id,)
+        ).fetchone()
+
+    def update_owner_plan(self, owner_id: str, plan: str) -> None:
+        with self._lock, self._conn as conn:
+            conn.execute("UPDATE url_shortener_owners SET plan = ? WHERE owner_id = ?", (plan, owner_id))
+
+    def code_exists(self, short_code: str) -> bool:
+        return self._conn.execute(
+            "SELECT 1 FROM url_shortener_links WHERE short_code = ?", (short_code,)
+        ).fetchone() is not None
+
+    def insert_link(self, **kwargs: Any) -> None:
+        with self._lock, self._conn as conn:
+            conn.execute("""
+                INSERT INTO url_shortener_links
+                    (short_code, long_url, owner_id, created_at, expires_at,
+                     max_clicks, click_count, last_accessed_at, is_active, tags, note,
+                     password_hash, utm_source, utm_medium, utm_campaign,
+                     campaign_id, custom_domain, fallback_url, webhook_url, geo_targeting)
+                VALUES
+                    (:short_code, :long_url, :owner_id, :created_at, :expires_at,
+                     :max_clicks, 0, NULL, 1, :tags, :note,
+                     :password_hash, :utm_source, :utm_medium, :utm_campaign,
+                     :campaign_id, :custom_domain, :fallback_url, :webhook_url,
+                     :geo_targeting)
+            """, kwargs)
+
+    def update_link(self, short_code: str, **updates: Any) -> None:
+        with self._lock, self._conn as conn:
+            sets = ", ".join(f"{k} = ?" for k in updates)
+            vals = list(updates.values()) + [short_code]
+            conn.execute(f"UPDATE url_shortener_links SET {sets} WHERE short_code = ?", vals)
+
+    def get_link(self, short_code: str) -> Optional[sqlite3.Row]:
+        return self._conn.execute(
+            "SELECT * FROM url_shortener_links WHERE short_code = ?", (short_code,)
+        ).fetchone()
+
+    def list_links_for_owner(self, owner_id: str) -> List[sqlite3.Row]:
+        return self._conn.execute(
+            "SELECT * FROM url_shortener_links WHERE owner_id = ? ORDER BY created_at DESC",
+            (owner_id,),
+        ).fetchall()
+
+    def deactivate_link(self, short_code: str) -> None:
+        with self._lock, self._conn as conn:
+            conn.execute("UPDATE url_shortener_links SET is_active = 0 WHERE short_code = ?", (short_code,))
+
+    def record_click(self, short_code: str, referrer: Optional[str] = None,
+                     user_agent: Optional[str] = None, ip_address: Optional[str] = None) -> None:
+        now = _now_iso()
+        parsed = _parse_user_agent(user_agent or "")
+        with self._lock, self._conn as conn:
+            conn.execute(
+                "INSERT INTO url_shortener_clicks "
+                "(short_code, referrer, user_agent, ip_address, country, browser, device, os, clicked_at) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (short_code, referrer, user_agent, ip_address, None,
+                 parsed["browser"], parsed["device"], parsed["os"], now),
+            )
+            conn.execute(
+                "UPDATE url_shortener_links SET click_count = click_count + 1, last_accessed_at = ? "
+                "WHERE short_code = ?",
+                (now, short_code),
+            )
+
+    def get_click_analytics(self, short_code: str) -> Dict[str, Any]:
+        total = self._conn.execute(
+            "SELECT COUNT(*) FROM url_shortener_clicks WHERE short_code = ?", (short_code,)
+        ).fetchone()[0]
+        browsers = dict(self._conn.execute(
+            "SELECT browser, COUNT(*) FROM url_shortener_clicks WHERE short_code = ? "
+            "AND browser IS NOT NULL GROUP BY browser ORDER BY COUNT(*) DESC",
+            (short_code,)
+        ).fetchall())
+        devices = dict(self._conn.execute(
+            "SELECT device, COUNT(*) FROM url_shortener_clicks WHERE short_code = ? "
+            "AND device IS NOT NULL GROUP BY device ORDER BY COUNT(*) DESC",
+            (short_code,)
+        ).fetchall())
+        referrers = dict(self._conn.execute(
+            "SELECT COALESCE(NULLIF(referrer, ''), '(direct)') AS ref, COUNT(*) AS n "
+            "FROM url_shortener_clicks WHERE short_code = ? "
+            "GROUP BY ref ORDER BY n DESC LIMIT 10",
+            (short_code,)
+        ).fetchall())
+        os_data = dict(self._conn.execute(
+            "SELECT os, COUNT(*) FROM url_shortener_clicks WHERE short_code = ? "
+            "AND os IS NOT NULL GROUP BY os ORDER BY COUNT(*) DESC",
+            (short_code,)
+        ).fetchall())
+        recent = self._conn.execute(
+            "SELECT clicked_at FROM url_shortener_clicks WHERE short_code = ? "
+            "ORDER BY clicked_at DESC LIMIT 50",
+            (short_code,)
+        ).fetchall()
+        return {
+            "total_clicks": total,
+            "browsers": browsers,
+            "devices": devices,
+            "operating_systems": os_data,
+            "top_referrers": referrers,
+            "recent_clicks": [r["clicked_at"] for r in recent],
+        }
+
+    def get_link_count_for_owner(self, owner_id: str) -> int:
+        return self._conn.execute(
+            "SELECT COUNT(*) FROM url_shortener_links WHERE owner_id = ?", (owner_id,)
+        ).fetchone()[0]
+
+    def audit(self, owner_id: Optional[str], action: str, short_code: Optional[str], detail: str = "") -> None:
+        with self._lock, self._conn as conn:
+            conn.execute(
+                "INSERT INTO url_shortener_audit_log (timestamp, owner_id, action, short_code, detail) "
+                "VALUES (?, ?, ?, ?, ?)",
+                (_now_iso(), owner_id, action, short_code, detail),
+            )
+
+    def export_links_csv(self, owner_id: str) -> str:
+        rows = self._conn.execute(
+            "SELECT short_code, long_url, created_at, click_count, is_active, tags, note "
+            "FROM url_shortener_links WHERE owner_id = ? ORDER BY created_at DESC", (owner_id,)
+        ).fetchall()
+        buf = io.StringIO()
+        w = csv.writer(buf)
+        w.writerow(["short_code", "long_url", "created_at", "click_count", "is_active", "tags", "note"])
+        for r in rows:
+            w.writerow([r["short_code"], r["long_url"], r["created_at"],
+                       r["click_count"], r["is_active"], r["tags"] or "", r["note"] or ""])
+        return buf.getvalue()
+
+    def export_clicks_csv(self, short_code: str) -> str:
+        rows = self._conn.execute(
+            "SELECT id, referrer, user_agent, ip_address, browser, device, os, clicked_at "
+            "FROM url_shortener_clicks WHERE short_code = ? ORDER BY clicked_at DESC", (short_code,)
+        ).fetchall()
+        buf = io.StringIO()
+        w = csv.writer(buf)
+        w.writerow(["id", "referrer", "user_agent", "ip_address", "browser", "device", "os", "clicked_at"])
+        for r in rows:
+            w.writerow([r["id"], r["referrer"], r["user_agent"], r["ip_address"],
+                       r["browser"], r["device"], r["os"], r["clicked_at"]])
+        return buf.getvalue()
+
+    def list_links_by_campaign(self, campaign_id: str) -> List[sqlite3.Row]:
+        return self._conn.execute(
+            "SELECT * FROM url_shortener_links WHERE campaign_id = ? ORDER BY created_at DESC",
+            (campaign_id,),
+        ).fetchall()
+
+    def create_campaign(self, campaign_id: str, owner_id: str, name: str, description: Optional[str] = None) -> None:
+        with self._lock, self._conn as conn:
+            conn.execute(
+                "INSERT INTO url_shortener_campaigns (campaign_id, owner_id, name, description, created_at) "
+                "VALUES (?, ?, ?, ?, ?)",
+                (campaign_id, owner_id, name, description, _now_iso()),
+            )
+
+    def get_campaign(self, campaign_id: str) -> Optional[sqlite3.Row]:
+        return self._conn.execute(
+            "SELECT * FROM url_shortener_campaigns WHERE campaign_id = ?", (campaign_id,)
+        ).fetchone()
+
+    def list_campaigns_for_owner(self, owner_id: str) -> List[sqlite3.Row]:
+        return self._conn.execute(
+            "SELECT * FROM url_shortener_campaigns WHERE owner_id = ? ORDER BY created_at DESC",
+            (owner_id,),
+        ).fetchall()
+
+    def deactivate_campaign(self, campaign_id: str) -> None:
+        with self._lock, self._conn as conn:
+            conn.execute(
+                "UPDATE url_shortener_campaigns SET is_active = 0 WHERE campaign_id = ?",
+                (campaign_id,),
+            )
+
+    def get_campaign_link_count(self, campaign_id: str) -> int:
+        return self._conn.execute(
+            "SELECT COUNT(*) FROM url_shortener_links WHERE campaign_id = ?", (campaign_id,)
+        ).fetchone()[0]
+
+    def get_campaign_total_clicks(self, campaign_id: str) -> int:
+        return self._conn.execute(
+            "SELECT COALESCE(SUM(click_count), 0) FROM url_shortener_links WHERE campaign_id = ?",
+            (campaign_id,),
+        ).fetchone()[0]
+
+    def get_campaign_analytics(self, campaign_id: str) -> Dict[str, Any]:
+        codes = self._conn.execute(
+            "SELECT short_code FROM url_shortener_links WHERE campaign_id = ?", (campaign_id,)
+        ).fetchall()
+        total = 0
+        browsers: Dict[str, int] = {}
+        devices: Dict[str, int] = {}
+        os_data: Dict[str, int] = {}
+        for (code,) in codes:
+            total += self._conn.execute(
+                "SELECT COUNT(*) FROM url_shortener_clicks WHERE short_code = ?", (code,)
+            ).fetchone()[0]
+            for b, n in self._conn.execute(
+                "SELECT browser, COUNT(*) FROM url_shortener_clicks WHERE short_code = ? "
+                "AND browser IS NOT NULL GROUP BY browser", (code,)
+            ).fetchall():
+                browsers[b] = browsers.get(b, 0) + n
+            for d, n in self._conn.execute(
+                "SELECT device, COUNT(*) FROM url_shortener_clicks WHERE short_code = ? "
+                "AND device IS NOT NULL GROUP BY device", (code,)
+            ).fetchall():
+                devices[d] = devices.get(d, 0) + n
+            for o, n in self._conn.execute(
+                "SELECT os, COUNT(*) FROM url_shortener_clicks WHERE short_code = ? "
+                "AND os IS NOT NULL GROUP BY os", (code,)
+            ).fetchall():
+                os_data[o] = os_data.get(o, 0) + n
+        return {"total_clicks": total, "browsers": browsers, "devices": devices, "operating_systems": os_data}
+
+    def owner_summary(self, owner_id: str) -> Dict[str, Any]:
+        row = self.get_owner(owner_id)
+        if row is None:
+            return {}
+        link_count = self.get_link_count_for_owner(owner_id)
+        total_clicks = self._conn.execute(
+            "SELECT COALESCE(SUM(click_count), 0) FROM url_shortener_links WHERE owner_id = ?",
+            (owner_id,)
+        ).fetchone()[0]
+        active = self._conn.execute(
+            "SELECT COUNT(*) FROM url_shortener_links WHERE owner_id = ? AND is_active = 1",
+            (owner_id,)
+        ).fetchone()[0]
+        return {
+            "owner_id": owner_id,
+            "name": row["name"],
+            "plan": row["plan"],
+            "total_links": link_count,
+            "active_links": active,
+            "total_clicks": total_clicks,
+            "created_at": row["created_at"],
+        }
+
+
+class URLShortenerService:
+    def __init__(self, db_path: Optional[str] = None):
+        self.db_path = db_path or os.environ.get("URL_SHORTENER_DB", "data/url_shortener.db")
+        self.storage = Storage(self.db_path)
+        logger.info("URLShortenerService initialized (db=%s)", self.db_path)
+
+    def close(self) -> None:
+        self.storage.close()
+
+    def _get_owner_plan(self, owner_id: str) -> Dict[str, Any]:
+        row = self.storage.get_owner(owner_id)
+        if row is None:
+            raise AuthenticationError(f"Owner '{owner_id}' does not exist")
+        return PLAN_LIMITS.get(row["plan"], PLAN_LIMITS["free"])
+
+    def create_owner(self, name: str, plan: str = "free") -> Dict[str, str]:
+        if plan not in PLAN_LIMITS:
+            plan = "free"
+        owner_id = secrets.token_hex(8)
+        api_key = f"sk_{secrets.token_urlsafe(24)}"
+        self.storage.create_owner(owner_id, name, _hash_key(api_key), plan)
+        self.storage.audit(owner_id, "CREATE_OWNER", None, f"plan={plan}")
+        logger.info("Created owner '%s' (%s) plan=%s", name, owner_id, plan)
+        return {"owner_id": owner_id, "api_key": api_key}
+
+    def authenticate(self, api_key: str) -> str:
+        row = self.storage.get_owner_by_key_hash(_hash_key(api_key))
+        if row is None:
+            raise AuthenticationError("Invalid API key")
+        return row["owner_id"]
+
+    def update_owner_plan(self, owner_id: str, plan: str) -> Dict[str, Any]:
+        if plan not in PLAN_LIMITS:
+            raise ValidationError(f"Invalid plan '{plan}'. Must be one of: {', '.join(PLAN_LIMITS)}")
+        row = self.storage.get_owner(owner_id)
+        if row is None:
+            raise AuthenticationError(f"Owner '{owner_id}' does not exist")
+        self.storage.update_owner_plan(owner_id, plan)
+        self.storage.audit(owner_id, "UPDATE_PLAN", None, f"new_plan={plan}")
+        logger.info("Owner %s plan updated to %s", owner_id, plan)
+        return {"owner_id": owner_id, "plan": plan}
+
+    def shorten(
+        self,
+        long_url: str,
+        owner_id: str,
+        custom_alias: Optional[str] = None,
+        expires_in_days: Optional[int] = None,
+        expires_at: Optional[datetime] = None,
+        max_clicks: Optional[int] = None,
+        tags: Optional[List[str]] = None,
+        note: Optional[str] = None,
+        password: Optional[str] = None,
+        utm_source: Optional[str] = None,
+        utm_medium: Optional[str] = None,
+        utm_campaign: Optional[str] = None,
+        campaign_id: Optional[str] = None,
+        custom_domain: Optional[str] = None,
+        fallback_url: Optional[str] = None,
+        webhook_url: Optional[str] = None,
+        geo_targeting: Optional[Dict[str, Any]] = None,
+    ) -> ShortLink:
+        plan_limits = self._get_owner_plan(owner_id)
+
+        if custom_alias is not None:
+            if not plan_limits["custom_alias"]:
+                raise PlanLimitExceededError("Your plan does not support custom aliases. Upgrade to Pro.")
+            validate_alias(custom_alias)
+
+        if max_clicks is not None and max_clicks < 1:
+            raise ValidationError("max_clicks must be a positive integer")
+
+        if expires_in_days is not None and expires_in_days < 1:
+            raise ValidationError("expires_in_days must be a positive integer")
+
+        if tags and not plan_limits["tags"]:
+            raise PlanLimitExceededError("Your plan does not support tags. Upgrade to Pro.")
+
+        validate_url(long_url)
+
+        current_count = self.storage.get_link_count_for_owner(owner_id)
+        if current_count >= plan_limits["max_links"]:
+            raise PlanLimitExceededError(
+                f"You have reached the limit of {plan_limits['max_links']} links on your plan. "
+                "Upgrade to create more."
+            )
+
+        if expires_at is None and expires_in_days is not None:
+            expires_at = datetime.now(timezone.utc) + timedelta(days=expires_in_days)
+
+        if tags is not None:
+            if len(tags) > 20:
+                raise ValidationError("At most 20 tags are allowed")
+            for tag in tags:
+                if not tag or len(tag) > 50:
+                    raise ValidationError("Each tag must be 1-50 characters")
+                if "," in tag:
+                    raise ValidationError("Tags must not contain commas")
+
+        if note is not None and len(note) > 2000:
+            raise ValidationError("note must not exceed 2000 characters")
+
+        if password is not None and len(password) > 128:
+            raise ValidationError("password must not exceed 128 characters")
+
+        if webhook_url and not plan_limits["webhooks"]:
+            raise PlanLimitExceededError("Your plan does not support webhooks. Upgrade to Enterprise.")
+        if webhook_url:
+            validate_url(webhook_url)
+
+        if custom_domain and not plan_limits["custom_domain"]:
+            raise PlanLimitExceededError("Your plan does not support custom domains. Upgrade to Enterprise.")
+
+        if geo_targeting and not plan_limits["geo_targeting"]:
+            raise PlanLimitExceededError("Your plan does not support geo-targeting. Upgrade to Enterprise.")
+
+        if campaign_id:
+            camp = self.storage.get_campaign(campaign_id)
+            if camp is None:
+                raise ValidationError(f"Campaign '{campaign_id}' not found")
+            if camp["owner_id"] != owner_id:
+                raise AuthorizationError("Campaign does not belong to this owner")
+
+        if fallback_url:
+            validate_url(fallback_url)
+
+        with self.storage.writer_lock():
+            code = custom_alias
+            if code is not None:
+                if self.storage.code_exists(code):
+                    raise AliasTakenError(f"Alias '{code}' is already in use")
+            else:
+                for _ in range(5):
+                    candidate = generate_short_code(7)
+                    if not self.storage.code_exists(candidate):
+                        code = candidate
+                        break
+                else:
+                    raise URLShortenerError("Could not generate a unique short code, try again")
+
+            self.storage.insert_link(
+                short_code=code,
+                long_url=long_url,
+                owner_id=owner_id,
+                created_at=_now_iso(),
+                expires_at=expires_at.isoformat() if expires_at else None,
+                max_clicks=max_clicks,
+                tags=",".join(tags) if tags else None,
+                note=note,
+                password_hash=_hash_password(password) if password else None,
+                utm_source=utm_source,
+                utm_medium=utm_medium,
+                utm_campaign=utm_campaign,
+                campaign_id=campaign_id,
+                custom_domain=custom_domain,
+                fallback_url=fallback_url,
+                webhook_url=webhook_url,
+                geo_targeting=json.dumps(geo_targeting) if geo_targeting else None,
+            )
+
+        self.storage.audit(owner_id, "CREATE_LINK", code, long_url)
+        logger.info("Owner %s created short link '%s'", owner_id, code)
+        return self._row_to_link(self.storage.get_link(code))
+
+    def bulk_shorten(self, owner_id: str, requests: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        results = []
+        for i, req in enumerate(requests):
+            try:
+                link = self.shorten(owner_id=owner_id, **req)
+                results.append({"index": i, "ok": True, "short_code": link.short_code, "long_url": link.long_url})
+            except URLShortenerError as e:
+                results.append({"index": i, "ok": False, "error": str(e)})
+        return results
+
+    def resolve(self, short_code: str, referrer: Optional[str] = None,
+                user_agent: Optional[str] = None, ip_address: Optional[str] = None) -> str:
+        with self.storage.writer_lock():
+            row = self.storage.get_link(short_code)
+            if row is None:
+                raise LinkNotFoundError(f"No link found for code '{short_code}'")
+
+            if row["password_hash"]:
+                raise LinkExpiredError(f"Link '{short_code}' is password-protected. Provide password via X-Link-Password header.")
+
+            fallback = row["fallback_url"]
+            if not row["is_active"]:
+                if fallback:
+                    return fallback
+                raise LinkExpiredError(f"Link '{short_code}' has been deactivated")
+
+            expires_at = _parse_iso(row["expires_at"])
+            if expires_at and datetime.now(timezone.utc) > expires_at:
+                if fallback:
+                    return fallback
+                raise LinkExpiredError(f"Link '{short_code}' expired on {expires_at.isoformat()}")
+
+            if row["max_clicks"] is not None and row["click_count"] >= row["max_clicks"]:
+                if fallback:
+                    return fallback
+                raise LinkExpiredError(f"Link '{short_code}' has reached its click limit")
+
+            self.storage.record_click(short_code, referrer, user_agent, ip_address)
+            long_url = row["long_url"]
+
+            # Geo-targeting: override destination based on geo/device rules
+            geo_raw = row["geo_targeting"]
+            if isinstance(geo_raw, str) and geo_raw:
+                try:
+                    rules = json.loads(geo_raw)
+                    long_url = _apply_geo_targeting(rules, long_url, ip_address, user_agent)
+                except (json.JSONDecodeError, Exception):
+                    pass
+
+            if row["utm_source"] or row["utm_medium"] or row["utm_campaign"]:
+                parsed = urlparse(long_url)
+                existing = dict(urlparse.parse_qsl(parsed.query))
+                if row["utm_source"]:
+                    existing["utm_source"] = row["utm_source"]
+                if row["utm_medium"]:
+                    existing["utm_medium"] = row["utm_medium"]
+                if row["utm_campaign"]:
+                    existing["utm_campaign"] = row["utm_campaign"]
+                new_qs = "&".join(f"{k}={v}" for k, v in existing.items())
+                long_url = parsed._replace(query=new_qs).geturl()
+
+        # Fire webhook asynchronously outside the lock
+        webhook = row["webhook_url"]
+        if webhook:
+            try:
+                _fire_click_webhook(webhook, short_code, row["long_url"], referrer, user_agent, ip_address)
+            except Exception:
+                logger.exception("Failed to fire webhook for %s", short_code)
+
+        logger.info("Resolved '%s'", short_code)
+        return long_url
+
+    def resolve_with_password(self, short_code: str, password: str, referrer: Optional[str] = None,
+                               user_agent: Optional[str] = None, ip_address: Optional[str] = None) -> str:
+        with self.storage.writer_lock():
+            row = self.storage.get_link(short_code)
+            if row is None:
+                raise LinkNotFoundError(f"No link found for code '{short_code}'")
+            if not row["password_hash"]:
+                return self.resolve(short_code, referrer, user_agent, ip_address)
+            if _hash_password(password) != row["password_hash"]:
+                raise AuthorizationError("Incorrect password")
+            if not row["is_active"]:
+                raise LinkExpiredError(f"Link '{short_code}' has been deactivated")
+            expires_at = _parse_iso(row["expires_at"])
+            if expires_at and datetime.now(timezone.utc) > expires_at:
+                raise LinkExpiredError(f"Link '{short_code}' expired on {expires_at.isoformat()}")
+            if row["max_clicks"] is not None and row["click_count"] >= row["max_clicks"]:
+                raise LinkExpiredError(f"Link '{short_code}' has reached its click limit")
+            self.storage.record_click(short_code, referrer, user_agent, ip_address)
+            return row["long_url"]
+
+    def list_links(self, owner_id: str) -> List[ShortLink]:
+        return [self._row_to_link(r) for r in self.storage.list_links_for_owner(owner_id)]
+
+    def update_link(self, short_code: str, owner_id: str, **updates: Any) -> ShortLink:
+        row = self.storage.get_link(short_code)
+        if row is None:
+            raise LinkNotFoundError(f"No link found for code '{short_code}'")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this link")
+
+        safe_updates: Dict[str, Any] = {}
+        if "long_url" in updates:
+            validate_url(updates["long_url"])
+            safe_updates["long_url"] = updates["long_url"]
+        if "note" in updates:
+            if updates["note"] is not None and len(updates["note"]) > 2000:
+                raise ValidationError("note must not exceed 2000 characters")
+            safe_updates["note"] = updates["note"]
+        if "tags" in updates:
+            tags = updates["tags"]
+            if tags is not None:
+                if len(tags) > 20:
+                    raise ValidationError("At most 20 tags are allowed")
+                for t in tags:
+                    if not t or len(t) > 50:
+                        raise ValidationError("Each tag must be 1-50 characters")
+                    if "," in t:
+                        raise ValidationError("Tags must not contain commas")
+                safe_updates["tags"] = ",".join(tags)
+            else:
+                safe_updates["tags"] = None
+        if "max_clicks" in updates:
+            if updates["max_clicks"] is not None and updates["max_clicks"] < 1:
+                raise ValidationError("max_clicks must be a positive integer")
+            safe_updates["max_clicks"] = updates["max_clicks"]
+        if "expires_at" in updates:
+            safe_updates["expires_at"] = updates["expires_at"].isoformat() if updates["expires_at"] else None
+        if "password" in updates:
+            safe_updates["password_hash"] = _hash_password(updates["password"]) if updates["password"] else None
+        if "webhook_url" in updates:
+            if updates["webhook_url"]:
+                validate_url(updates["webhook_url"])
+            safe_updates["webhook_url"] = updates["webhook_url"]
+        if "fallback_url" in updates:
+            if updates["fallback_url"]:
+                validate_url(updates["fallback_url"])
+            safe_updates["fallback_url"] = updates["fallback_url"]
+        if "campaign_id" in updates:
+            safe_updates["campaign_id"] = updates["campaign_id"]
+        if "custom_domain" in updates:
+            safe_updates["custom_domain"] = updates["custom_domain"]
+        if "geo_targeting" in updates:
+            safe_updates["geo_targeting"] = json.dumps(updates["geo_targeting"]) if updates["geo_targeting"] else None
+
+        if safe_updates:
+            self.storage.update_link(short_code, **safe_updates)
+            detail = "; ".join(f"{k}={v}" for k, v in safe_updates.items())
+            self.storage.audit(owner_id, "UPDATE_LINK", short_code, detail)
+            logger.info("Owner %s updated link '%s'", owner_id, short_code)
+
+        return self._row_to_link(self.storage.get_link(short_code))
+
+    def deactivate_link(self, short_code: str, owner_id: str) -> None:
+        row = self.storage.get_link(short_code)
+        if row is None:
+            raise LinkNotFoundError(f"No link found for code '{short_code}'")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this link")
+        self.storage.deactivate_link(short_code)
+        self.storage.audit(owner_id, "DEACTIVATE_LINK", short_code)
+        logger.info("Owner %s deactivated link '%s'", owner_id, short_code)
+
+    def get_link_stats(self, short_code: str, owner_id: str) -> Dict[str, Any]:
+        row = self.storage.get_link(short_code)
+        if row is None:
+            raise LinkNotFoundError(f"No link found for code '{short_code}'")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this link")
+        stats = self.storage.get_click_analytics(short_code)
+        return {
+            "short_code": row["short_code"],
+            "long_url": row["long_url"],
+            "click_count": row["click_count"],
+            "created_at": row["created_at"],
+            "last_accessed_at": row["last_accessed_at"],
+            "is_active": bool(row["is_active"]),
+            "expires_at": row["expires_at"],
+            "max_clicks": row["max_clicks"],
+            "has_password": bool(row["password_hash"]),
+            "utm": {
+                "source": row["utm_source"],
+                "medium": row["utm_medium"],
+                "campaign": row["utm_campaign"],
+            },
+            "analytics": stats,
+            "campaign_id": row["campaign_id"],
+            "custom_domain": row["custom_domain"],
+            "fallback_url": row["fallback_url"],
+            "webhook_url": row["webhook_url"],
+        }
+
+    def get_click_details(self, short_code: str, owner_id: str, limit: int = 100) -> List[Dict[str, Any]]:
+        row = self.storage.get_link(short_code)
+        if row is None:
+            raise LinkNotFoundError(f"No link found for code '{short_code}'")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this link")
+        import sqlite3 as _sqlite3
+        rows = self.storage._conn.execute(
+            "SELECT id, referrer, user_agent, ip_address, browser, device, os, clicked_at "
+            "FROM url_shortener_clicks WHERE short_code = ? ORDER BY clicked_at DESC LIMIT ?",
+            (short_code, limit),
+        ).fetchall()
+        return [dict(r) for r in rows]
+
+    def get_qr_code(self, short_code: str) -> Dict[str, str]:
+        row = self.storage.get_link(short_code)
+        if row is None:
+            raise LinkNotFoundError(f"No link found for code '{short_code}'")
+        short_url = f"{SRV_BASE_URL}/url-shortener/{short_code}"
+        return {
+            "short_code": short_code,
+            "short_url": short_url,
+            "qr_image_url": _build_qr_data_url(short_code),
+        }
+
+    def export_links(self, owner_id: str) -> str:
+        return self.storage.export_links_csv(owner_id)
+
+    def export_clicks(self, short_code: str, owner_id: str) -> str:
+        row = self.storage.get_link(short_code)
+        if row is None:
+            raise LinkNotFoundError(f"No link found for code '{short_code}'")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this link")
+        return self.storage.export_clicks_csv(short_code)
+
+    def owner_summary(self, owner_id: str) -> Dict[str, Any]:
+        return self.storage.owner_summary(owner_id)
+
+    def create_campaign(self, owner_id: str, name: str, description: Optional[str] = None) -> Dict[str, Any]:
+        plan_limits = self._get_owner_plan(owner_id)
+        if not plan_limits.get("campaigns", False):
+            raise PlanLimitExceededError("Your plan does not support campaigns. Upgrade to Pro or Enterprise.")
+        campaign_id = secrets.token_hex(8)
+        self.storage.create_campaign(campaign_id, owner_id, name, description)
+        self.storage.audit(owner_id, "CREATE_CAMPAIGN", None, f"name={name}")
+        logger.info("Owner %s created campaign '%s'", owner_id, campaign_id)
+        return {"campaign_id": campaign_id, "name": name, "description": description}
+
+    def list_campaigns(self, owner_id: str) -> List[Dict[str, Any]]:
+        rows = self.storage.list_campaigns_for_owner(owner_id)
+        result = []
+        for r in rows:
+            link_count = self.storage.get_campaign_link_count(r["campaign_id"])
+            total_clicks = self.storage.get_campaign_total_clicks(r["campaign_id"])
+            result.append({
+                "campaign_id": r["campaign_id"],
+                "name": r["name"],
+                "description": r["description"],
+                "created_at": r["created_at"],
+                "is_active": bool(r["is_active"]),
+                "link_count": link_count,
+                "total_clicks": total_clicks,
+            })
+        return result
+
+    def get_campaign(self, campaign_id: str, owner_id: str) -> Dict[str, Any]:
+        row = self.storage.get_campaign(campaign_id)
+        if row is None:
+            raise LinkNotFoundError(f"Campaign '{campaign_id}' not found")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this campaign")
+        link_count = self.storage.get_campaign_link_count(campaign_id)
+        total_clicks = self.storage.get_campaign_total_clicks(campaign_id)
+        analytics = self.storage.get_campaign_analytics(campaign_id)
+        return {
+            "campaign_id": row["campaign_id"],
+            "name": row["name"],
+            "description": row["description"],
+            "created_at": row["created_at"],
+            "is_active": bool(row["is_active"]),
+            "link_count": link_count,
+            "total_clicks": total_clicks,
+            "analytics": analytics,
+        }
+
+    def deactivate_campaign(self, campaign_id: str, owner_id: str) -> None:
+        row = self.storage.get_campaign(campaign_id)
+        if row is None:
+            raise LinkNotFoundError(f"Campaign '{campaign_id}' not found")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this campaign")
+        self.storage.deactivate_campaign(campaign_id)
+        self.storage.audit(owner_id, "DEACTIVATE_CAMPAIGN", None, campaign_id)
+        logger.info("Owner %s deactivated campaign '%s'", owner_id, campaign_id)
+
+    def list_campaign_links(self, campaign_id: str, owner_id: str) -> List[ShortLink]:
+        row = self.storage.get_campaign(campaign_id)
+        if row is None:
+            raise LinkNotFoundError(f"Campaign '{campaign_id}' not found")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this campaign")
+        return [self._row_to_link(r) for r in self.storage.list_links_by_campaign(campaign_id)]
+
+    def check_link_health(self, short_code: str, owner_id: str) -> Dict[str, Any]:
+        row = self.storage.get_link(short_code)
+        if row is None:
+            raise LinkNotFoundError(f"No link found for code '{short_code}'")
+        if row["owner_id"] != owner_id:
+            raise AuthorizationError("You do not own this link")
+        url = row["long_url"]
+        result = {"short_code": short_code, "long_url": url, "reachable": False, "status_code": None, "error": None}
+        try:
+            import httpx
+            resp = httpx.get(url, timeout=10.0, follow_redirects=True)
+            result["reachable"] = resp.is_success
+            result["status_code"] = resp.status_code
+        except ImportError:
+            import urllib.request
+            try:
+                resp = urllib.request.urlopen(url, timeout=10)
+                result["reachable"] = True
+                result["status_code"] = resp.getcode()
+            except Exception as exc:
+                result["error"] = str(exc)
+        except Exception as exc:
+            result["error"] = str(exc)
+        return result
+
+    def _row_to_link(self, row: sqlite3.Row) -> ShortLink:
+        gt_raw = row["geo_targeting"]
+        geo = json.loads(gt_raw) if isinstance(gt_raw, str) and gt_raw else None
+        return ShortLink(
+            short_code=row["short_code"],
+            long_url=row["long_url"],
+            owner_id=row["owner_id"],
+            created_at=_parse_iso(row["created_at"]),
+            expires_at=_parse_iso(row["expires_at"]),
+            max_clicks=row["max_clicks"],
+            click_count=row["click_count"],
+            is_active=bool(row["is_active"]),
+            tags=row["tags"].split(",") if row["tags"] else [],
+            note=row["note"],
+            password_hash=row["password_hash"],
+            utm_source=row["utm_source"],
+            utm_medium=row["utm_medium"],
+            utm_campaign=row["utm_campaign"],
+            campaign_id=row["campaign_id"],
+            custom_domain=row["custom_domain"],
+            fallback_url=row["fallback_url"],
+            webhook_url=row["webhook_url"],
+            geo_targeting=geo,
+        )