feat(shortener): public redirect + async click tracking + QR + analytics (B3-B4)

services/flask-backend/app/__init__.py

@@ -269,6 +269,16 @@ def _register_blueprints(app: Quart) -> None:
     app.register_blueprint(urls_bp, url_prefix="/api/v1")
     app.register_blueprint(collections_bp, url_prefix="/api/v1")
 
+    # Redirect endpoint (public, at root — NO auth required)
+    from .redirect import redirect_bp
+
+    app.register_blueprint(redirect_bp)
+
+    # Analytics endpoints (Phase 3 — auth required)
+    from .analytics import analytics_bp
+
+    app.register_blueprint(analytics_bp)
+
     # OAuth2 token endpoints (Phase 2)
     from .oauth import oauth_bp
 

services/flask-backend/app/analytics.py

@@ -0,0 +1,374 @@
+"""
+Analytics Endpoints for URL Shortener.
+
+Provides click analytics with distinction between BASIC and ADVANCED.
+All endpoints are tenant-scoped.
+
+BASIC (Free/Pro):
+  - Total clicks
+  - Click timeline (clicks per day)
+  - Top referers list
+
+ADVANCED (Enterprise — TODO(tiering)):
+  - Geo breakdown (country/region/city)
+  - Device type breakdown
+  - Browser breakdown
+  - OS breakdown
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from dataclasses import dataclass
+
+from quart import Blueprint, g, jsonify, request
+from werkzeug.exceptions import BadRequest, NotFound
+
+from .auth import auth_required
+from .models import get_db
+from .rbac import require_scope
+
+analytics_bp = Blueprint("analytics", __name__, url_prefix="/api/v1/analytics")
+
+
+@dataclass(slots=True)
+class ClickDayBucket:
+    """Daily click bucket."""
+
+    date: str
+    clicks: int
+
+
+@dataclass(slots=True)
+class RefererStat:
+    """Referer statistic."""
+
+    referer: str
+    clicks: int
+
+
+@dataclass(slots=True)
+class PerLinkAnalytics:
+    """Per-link analytics response."""
+
+    url_id: int
+    short_code: str
+    long_url: str
+    total_clicks: int
+    clicks_by_day: list[ClickDayBucket]
+    top_referers: list[RefererStat]
+
+
+@dataclass(slots=True)
+class TenantAnalyticsSummary:
+    """Tenant-wide analytics summary."""
+
+    total_links: int
+    total_clicks: int
+    top_links: list[dict]
+    clicks_by_day: list[ClickDayBucket]
+
+
+@analytics_bp.route("/urls/<int:url_id>", methods=["GET"])
+@auth_required
+@require_scope("analytics:read")
+async def get_url_analytics(url_id: int):
+    """
+    Get analytics for a specific URL.
+
+    Query params:
+    - from: ISO date (YYYY-MM-DD) to filter clicks from this date onwards
+    - to: ISO date (YYYY-MM-DD) to filter clicks up to this date
+
+    BASIC analytics:
+      - Total clicks
+      - Clicks by day (time series)
+      - Top referers
+
+    TODO(tiering): ADVANCED analytics
+      - Geo breakdown (country/region/city)
+      - Device type breakdown
+      - Browser breakdown
+      - OS breakdown
+
+    Returns:
+      {
+        "data": {
+          "url_id": 123,
+          "short_code": "abc123",
+          "long_url": "https://example.com",
+          "total_clicks": 42,
+          "clicks_by_day": [
+            {"date": "2025-01-20", "clicks": 10},
+            {"date": "2025-01-21", "clicks": 32}
+          ],
+          "top_referers": [
+            {"referer": "https://twitter.com", "clicks": 25},
+            {"referer": "direct", "clicks": 17}
+          ]
+        }
+      }
+
+    Requires: analytics:read scope (tenant-scoped)
+    """
+    db = get_db()
+    tenant_id = g.current_user.get("tenant_id")
+
+    if not tenant_id:
+        default_tenant = db(db.tenants.is_default == True).select().first()  # noqa
+        if not default_tenant:
+            raise BadRequest("No tenant found for user")
+        tenant_id = default_tenant.id
+
+    # Verify URL exists and belongs to tenant
+    url = (
+        db((db.urls.id == url_id) & (db.urls.tenant_id == tenant_id))
+        .select()
+        .first()
+    )
+    if not url:
+        raise NotFound("URL not found")
+
+    # Parse date filters
+    from_date = None
+    to_date = None
+
+    from_str = request.args.get("from")
+    if from_str:
+        try:
+            from_date = datetime.fromisoformat(from_str).replace(tzinfo=timezone.utc)
+        except ValueError:
+            raise BadRequest("Invalid 'from' date format (use YYYY-MM-DD)")
+
+    to_str = request.args.get("to")
+    if to_str:
+        try:
+            to_date = datetime.fromisoformat(to_str).replace(tzinfo=timezone.utc)
+        except ValueError:
+            raise BadRequest("Invalid 'to' date format (use YYYY-MM-DD)")
+
+    # Query clicks for this URL with date filters
+    query = db.link_clicks.url_id == url_id
+    if from_date:
+        query &= db.link_clicks.clicked_at >= from_date
+    if to_date:
+        query &= db.link_clicks.clicked_at <= to_date
+
+    clicks = db(query).select().as_list()
+
+    # BASIC ANALYTICS
+    total_clicks = len(clicks)
+
+    # Clicks by day
+    clicks_by_day_map: dict[str, int] = {}
+    for click in clicks:
+        if click["clicked_at"]:
+            day = click["clicked_at"].strftime("%Y-%m-%d")
+            clicks_by_day_map[day] = clicks_by_day_map.get(day, 0) + 1
+
+    clicks_by_day = [
+        ClickDayBucket(date=day, clicks=count)
+        for day, count in sorted(clicks_by_day_map.items())
+    ]
+
+    # Top referers (basic)
+    referer_map: dict[str | None, int] = {}
+    for click in clicks:
+        referer = click.get("referer") or "direct"
+        referer_map[referer] = referer_map.get(referer, 0) + 1
+
+    top_referers = sorted(
+        [RefererStat(referer=r, clicks=c) for r, c in referer_map.items()],
+        key=lambda x: x.clicks,
+        reverse=True,
+    )[:10]  # Top 10
+
+    analytics = PerLinkAnalytics(
+        url_id=url.id,
+        short_code=url.short_code,
+        long_url=url.long_url,
+        total_clicks=total_clicks,
+        clicks_by_day=clicks_by_day,
+        top_referers=top_referers,
+    )
+
+    # Convert dataclass to dict (with nested dataclass conversion)
+    response_data = {
+        "url_id": analytics.url_id,
+        "short_code": analytics.short_code,
+        "long_url": analytics.long_url,
+        "total_clicks": analytics.total_clicks,
+        "clicks_by_day": [
+            {"date": cdb.date, "clicks": cdb.clicks} for cdb in analytics.clicks_by_day
+        ],
+        "top_referers": [
+            {"referer": rs.referer, "clicks": rs.clicks} for rs in analytics.top_referers
+        ],
+    }
+
+    return jsonify({"data": response_data}), 200
+
+
+@analytics_bp.route("/summary", methods=["GET"])
+@auth_required
+@require_scope("analytics:read")
+async def get_tenant_analytics_summary():
+    """
+    Get analytics summary for all URLs in the tenant.
+
+    Query params:
+    - from: ISO date (YYYY-MM-DD)
+    - to: ISO date (YYYY-MM-DD)
+    - limit: Number of top links to return (default 10, max 50)
+    - offset: Pagination offset for top links (default 0)
+
+    BASIC analytics:
+      - Total links in tenant
+      - Total clicks across all links
+      - Top N links by clicks
+      - Clicks by day (time series)
+
+    TODO(tiering): ADVANCED analytics
+      - Geo breakdown
+      - Device type breakdown
+      - Browser breakdown
+      - OS breakdown
+
+    Returns:
+      {
+        "data": {
+          "total_links": 50,
+          "total_clicks": 1234,
+          "top_links": [
+            {"url_id": 1, "short_code": "abc", "clicks": 100},
+            ...
+          ],
+          "clicks_by_day": [
+            {"date": "2025-01-20", "clicks": 50},
+            ...
+          ]
+        }
+      }
+
+    Requires: analytics:read scope (tenant-scoped)
+    """
+    db = get_db()
+    tenant_id = g.current_user.get("tenant_id")
+
+    if not tenant_id:
+        default_tenant = db(db.tenants.is_default == True).select().first()  # noqa
+        if not default_tenant:
+            raise BadRequest("No tenant found for user")
+        tenant_id = default_tenant.id
+
+    # Parse date filters and pagination
+    from_date = None
+    to_date = None
+
+    from_str = request.args.get("from")
+    if from_str:
+        try:
+            from_date = datetime.fromisoformat(from_str).replace(tzinfo=timezone.utc)
+        except ValueError:
+            raise BadRequest("Invalid 'from' date format (use YYYY-MM-DD)")
+
+    to_str = request.args.get("to")
+    if to_str:
+        try:
+            to_date = datetime.fromisoformat(to_str).replace(tzinfo=timezone.utc)
+        except ValueError:
+            raise BadRequest("Invalid 'to' date format (use YYYY-MM-DD)")
+
+    try:
+        limit = int(request.args.get("limit", 10))
+        offset = int(request.args.get("offset", 0))
+        limit = min(limit, 50)  # Cap at 50
+    except ValueError:
+        raise BadRequest("Invalid limit/offset")
+
+    # Get all URLs for tenant
+    urls = db(db.urls.tenant_id == tenant_id).select().as_list()
+    total_links = len(urls)
+
+    # Get all clicks for this tenant's URLs within date range
+    url_ids = [u["id"] for u in urls]
+    if not url_ids:
+        # Empty tenant
+        return (
+            jsonify(
+                {
+                    "data": {
+                        "total_links": 0,
+                        "total_clicks": 0,
+                        "top_links": [],
+                        "clicks_by_day": [],
+                    }
+                }
+            ),
+            200,
+        )
+
+    # Build query for clicks
+    from pydal.objects import Query
+
+    query: Query = db.link_clicks.url_id.belongs(url_ids)
+    if from_date:
+        query &= db.link_clicks.clicked_at >= from_date
+    if to_date:
+        query &= db.link_clicks.clicked_at <= to_date
+
+    all_clicks = db(query).select().as_list()
+    total_clicks = len(all_clicks)
+
+    # Clicks by day (BASIC)
+    clicks_by_day_map: dict[str, int] = {}
+    for click in all_clicks:
+        if click["clicked_at"]:
+            day = click["clicked_at"].strftime("%Y-%m-%d")
+            clicks_by_day_map[day] = clicks_by_day_map.get(day, 0) + 1
+
+    clicks_by_day = [
+        ClickDayBucket(date=day, clicks=count)
+        for day, count in sorted(clicks_by_day_map.items())
+    ]
+
+    # Top links by clicks (BASIC)
+    clicks_per_url: dict[int, int] = {}
+    for click in all_clicks:
+        url_id = click["url_id"]
+        clicks_per_url[url_id] = clicks_per_url.get(url_id, 0) + 1
+
+    # Map back to URL metadata and sort
+    top_links_data = []
+    for url in urls:
+        url_id = url["id"]
+        clicks = clicks_per_url.get(url_id, 0)
+        top_links_data.append(
+            {
+                "url_id": url_id,
+                "short_code": url["short_code"],
+                "long_url": url["long_url"],
+                "clicks": clicks,
+            }
+        )
+
+    top_links_data.sort(key=lambda x: x["clicks"], reverse=True)
+    top_links = top_links_data[offset : offset + limit]
+
+    analytics = TenantAnalyticsSummary(
+        total_links=total_links,
+        total_clicks=total_clicks,
+        top_links=top_links,
+        clicks_by_day=clicks_by_day,
+    )
+
+    response_data = {
+        "total_links": analytics.total_links,
+        "total_clicks": analytics.total_clicks,
+        "top_links": analytics.top_links,
+        "clicks_by_day": [
+            {"date": cdb.date, "clicks": cdb.clicks} for cdb in analytics.clicks_by_day
+        ],
+    }
+
+    return jsonify({"data": response_data}), 200