fix: scope the self-link check to the destination host

config.py

@@ -241,6 +241,9 @@ class ClickEventsSettings(BaseSettings):
     # a message is dead-lettered to dlq_stream.
     claim_idle_ms: int = Field(default=60_000, ge=1000)
     max_deliveries: int = Field(default=5, ge=1)
+    # DLQ bound — sized for hours of clicks (~600B/event → 100k ≈ 60MB);
+    # recover with scripts/replay_dlq.py.
+    dlq_maxlen: int = Field(default=100_000, ge=1000)
     stats_interval_seconds: float = Field(default=30.0, gt=0)
 
     # Which consumer groups this worker process runs. Lets a future

repositories/indexes.py

@@ -64,7 +64,11 @@ async def ensure_indexes(
             timeseries={
                 "timeField": "clicked_at",
                 "metaField": "meta",
-                "granularity": "seconds",
+                # "hours" spans buckets over 30 days. Most links get sparse
+                # clicks, so finer granularity degenerates into one bucket
+                # per click. Existing deploys need a one-time collMod to
+                # match (the transition is one-way, coarser only).
+                "granularity": "hours",
             },
         )
     except (CollectionInvalid, OperationFailure) as e:

schemas/models/click.py

@@ -60,3 +60,14 @@ class ClickDoc(MongoBaseModel):
     utm_source: str | None = None
     utm_medium: str | None = None
     utm_campaign: str | None = None
+
+    def to_mongo(self) -> dict:
+        """Serialise for insertion, omitting None fields entirely.
+
+        Time-series buckets track one BSON type per field. An explicit
+        null following a string value (or vice versa) is a type conflict
+        that closes the bucket early, while an absent field packs fine.
+        Queries are unaffected: MongoDB treats missing and null the same
+        in $eq/$group, and the stats sentinels already map both.
+        """
+        return self.model_dump(by_alias=True, exclude_none=True)

scripts/replay_dlq.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env -S uv run --script
+# /// script
+# requires-python = ">=3.11"
+# dependencies = [
+#     "redis>=5",
+# ]
+# ///
+"""Replay dead-lettered click events back onto the main stream.
+
+Standalone — reads ``CLICK_EVENTS_QUEUE_REDIS_URI`` from the environment.
+Replayed events fan out to every consumer group again (streams have no
+per-group publish); replayed entries leave the DLQ unless ``--keep``.
+
+Usage::
+
+    uv run --env-file .env.production scripts/replay_dlq.py --dry-run
+    uv run --env-file .env.production scripts/replay_dlq.py --limit 500 --keep
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+
+from redis import Redis
+
+STREAM_FIELD_DATA = "__data__"  # payload field the workers' parser consumes
+DLQ_FIELD_SOURCE_ID = "dlq_source_id"
+DLQ_FIELD_GROUP = "dlq_group"
+DLQ_FIELD_REASON = "dlq_reason"
+
+_BATCH = 200
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Show what would be replayed without writing.",
+    )
+    parser.add_argument(
+        "--limit", type=int, default=0, help="Replay at most N entries (0 = all)."
+    )
+    parser.add_argument(
+        "--keep",
+        action="store_true",
+        help="Do not delete replayed entries from the DLQ.",
+    )
+    parser.add_argument("--stream", default="events:clicks")
+    parser.add_argument("--dlq-stream", default="events:clicks:dlq")
+    args = parser.parse_args()
+
+    uri = os.environ.get("CLICK_EVENTS_QUEUE_REDIS_URI")
+    if not uri:
+        sys.exit("CLICK_EVENTS_QUEUE_REDIS_URI not set in environment.")
+
+    redis = Redis.from_url(uri, decode_responses=True)
+    total = redis.xlen(args.dlq_stream)
+    print(f"DLQ {args.dlq_stream}: {total} entries")
+    if total == 0:
+        return
+
+    replayed = 0
+    by_group: dict[str, int] = {}
+    skipped = 0
+    cursor = "-"
+    while True:
+        entries = redis.xrange(args.dlq_stream, min=cursor, max="+", count=_BATCH)
+        if not entries:
+            break
+        for entry_id, fields in entries:
+            if args.limit and replayed >= args.limit:
+                break
+            data = fields.get(STREAM_FIELD_DATA)
+            if data is None:
+                skipped += 1
+                continue
+            group = fields.get(DLQ_FIELD_GROUP, "?")
+            if args.dry_run:
+                print(
+                    f"  would replay {entry_id} "
+                    f"(source={fields.get(DLQ_FIELD_SOURCE_ID, '?')}, "
+                    f"group={group}, reason={fields.get(DLQ_FIELD_REASON, '?')})"
+                )
+            else:
+                redis.xadd(args.stream, {STREAM_FIELD_DATA: data})
+                if not args.keep:
+                    redis.xdel(args.dlq_stream, entry_id)
+            replayed += 1
+            by_group[group] = by_group.get(group, 0) + 1
+        if (args.limit and replayed >= args.limit) or len(entries) < _BATCH:
+            break
+        # xrange min is inclusive — nudge past the last-seen id
+        cursor = f"({entries[-1][0]}"
+
+    verb = "would replay" if args.dry_run else "replayed"
+    print(
+        f"{verb} {replayed} (skipped {skipped} without payload) — by group: {by_group}"
+    )
+
+
+if __name__ == "__main__":
+    main()

shared/validators.py

@@ -19,6 +19,23 @@
 _ALLOWED_URL_SCHEMES = frozenset({"http", "https"})
 
 
+def is_self_referential(host: str | None, blocked_self_domains: Sequence[str]) -> bool:
+    """Return True when *host* is one of *blocked_self_domains* or a subdomain.
+
+    Host-scoped on purpose. A destination is only a redirect loop when the
+    request would come back to us, which is decided by the host alone — a
+    blocked name appearing in a path or query string ("?filter=spoo.me") is
+    someone else's URL that happens to mention us.
+    """
+    if not host:
+        return False
+    host = host.lower().rstrip(".")
+    return any(
+        host == domain or host.endswith(f".{domain}")
+        for domain in (d.lower().strip().rstrip(".") for d in blocked_self_domains)
+    )
+
+
 def validate_url(
     url: str,
     blocked_self_domains: Sequence[str] = ("spoo.me",),
@@ -27,18 +44,28 @@ def validate_url(
 
     Args:
         url: The URL string to validate.
-        blocked_self_domains: Bare hostnames whose substring presence in the
-            URL marks it as self-referential. Defaults to ``("spoo.me",)``
-            to prevent redirect loops.
+        blocked_self_domains: Bare hostnames that mark the URL as
+            self-referential when they are the destination's HOST (or a
+            parent of it). Defaults to ``("spoo.me",)`` to prevent redirect
+            loops. Matching is host-scoped: a destination that merely
+            mentions the name in its path or query is not a loop.
     """
+    # urlparse raises on a malformed IPv6 authority ("https://[::1"), and
+    # long_url reaches here as a plain str, so an unguarded parse escapes as
+    # a 500 on the shorten endpoints instead of a 400.
+    try:
+        parsed = urlparse(url)
+    except ValueError:
+        return False
     # Scheme allowlist defends against ftp/file/data/etc even if the
     # validators package widens its accepted schemes upstream.
-    if urlparse(url).scheme not in _ALLOWED_URL_SCHEMES:
+    if parsed.scheme not in _ALLOWED_URL_SCHEMES:
         return False
     if not _validators.url(url, skip_ipv4_addr=True, skip_ipv6_addr=True):
         return False
-    url_lower = url.lower()
-    return not any(domain in url_lower for domain in blocked_self_domains)
+    # hostname is already lowercased and strips any userinfo/port, so
+    # "https://spoo.me:443@evil.com" can't smuggle the check either way.
+    return not is_self_referential(parsed.hostname, blocked_self_domains)
 
 
 def validate_url_password(password: str, min_length: int = 8) -> bool:

tests/unit/repositories/test_indexes.py

@@ -140,7 +140,7 @@ async def test_ensure_indexes_creates_timeseries_collection(self):
             timeseries={
                 "timeField": "clicked_at",
                 "metaField": "meta",
-                "granularity": "seconds",
+                "granularity": "hours",
             },
         )
 

tests/unit/schemas/models/test_click.py

@@ -50,3 +50,19 @@ def test_to_mongo_round_trip(self):
         restored = ClickDoc.from_mongo(doc.to_mongo())
         assert restored.referrer == "google.com"
         assert restored.redirect_ms == doc.redirect_ms
+
+    def test_to_mongo_omits_none_fields(self):
+        # Explicit nulls flip the per-field BSON type and close time-series
+        # buckets; None fields must be absent from the insert document.
+        data = self._make(referrer=None, bot_name=None).to_mongo()
+        assert "referrer" not in data
+        assert "bot_name" not in data
+        assert "device" not in data
+        assert "utm_source" not in data
+        assert "domain" not in data["meta"]
+
+    def test_to_mongo_keeps_set_fields(self):
+        data = self._make(referrer="google.com", device="mobile").to_mongo()
+        assert data["referrer"] == "google.com"
+        assert data["device"] == "mobile"
+        assert data["country"] == "Unknown"

tests/unit/services/test_click_service.py

@@ -303,10 +303,11 @@ async def test_utm_tags_default_to_none(self):
 
         await handler.handle(make_context(url_data))
 
+        # Absent, not null: explicit nulls close time-series buckets.
         doc = d.click_repo.insert.call_args[0][0]
-        assert doc["utm_source"] is None
-        assert doc["utm_medium"] is None
-        assert doc["utm_campaign"] is None
+        assert "utm_source" not in doc
+        assert "utm_medium" not in doc
+        assert "utm_campaign" not in doc
 
     @pytest.mark.asyncio
     async def test_meta_carries_domain_from_cache(self):
@@ -320,17 +321,18 @@ async def test_meta_carries_domain_from_cache(self):
         assert doc["meta"]["domain"] == "links.acme.com"
 
     @pytest.mark.asyncio
-    async def test_meta_domain_none_when_cache_empty_string(self):
-        # Older cached entries pre-PR1 have domain="". Coerce to None so
-        # per-domain queries can distinguish "unknown" from a real value.
+    async def test_meta_domain_omitted_when_cache_empty_string(self):
+        # Older cached entries pre-PR1 have domain="". Coerced to None and
+        # then omitted from the insert doc ($eq: null matches missing, so
+        # per-domain queries still distinguish "unknown" from a real value).
         d = make_deps()
         handler = make_v2_handler(d.click_repo, d.url_repo, d.geoip, d.url_cache)
         url_data = make_v2_cache(domain="")
 
         await handler.handle(make_context(url_data))
 
         doc = d.click_repo.insert.call_args[0][0]
-        assert doc["meta"]["domain"] is None
+        assert "domain" not in doc["meta"]
 
     @pytest.mark.asyncio
     async def test_blocked_bot_skips_analytics_no_error(self):

tests/unit/shared/test_validators.py

@@ -21,16 +21,37 @@
         ("https://example.com/foo/bar?q=1", True),
         ("https://spoo.me/abc", False),
         ("https://SPOO.ME/abc", False),  # case-insensitive block
+        ("https://www.spoo.me/abc", False),  # subdomain of a blocked host
         ("not-a-url", False),
         ("http://192.168.1.1/path", False),  # IPv4 skipped by validator
+        ("https://[::1", False),  # urlparse raises on this, must not escape
+        # Host-scoped: a foreign destination that merely mentions the blocked
+        # name in its path, query or fragment is not a redirect loop. The
+        # dashboard hit this shortening a PostHog analytics URL filtered on
+        # spoo.me, which the old substring check rejected.
+        ("https://eu.posthog.com/project/1?filter=spoo.me", True),
+        ("https://example.com/spoo.me/guide", True),
+        ("https://example.com/#spoo.me", True),
+        ("https://notspoo.me/abc", True),  # suffix must be a label boundary
+        # userinfo can't smuggle either direction — hostname wins.
+        ("https://spoo.me@example.com/", True),
+        ("https://example.com@spoo.me/", False),
     ],
     ids=[
         "valid",
         "valid_with_path",
         "self_ref",
         "self_ref_uppercase",
+        "self_ref_subdomain",
         "plain_text",
         "ipv4",
+        "malformed_ipv6_authority",
+        "foreign_host_mentions_in_query",
+        "foreign_host_mentions_in_path",
+        "foreign_host_mentions_in_fragment",
+        "lookalike_host_not_blocked",
+        "userinfo_lookalike_allowed",
+        "userinfo_cannot_mask_self_host",
     ],
 )
 def test_validate_url(url, expected):

tests/unit/workers/test_dlq_guard.py

@@ -61,6 +61,18 @@ async def test_past_budget_moves_to_dlq(self):
         assert kwargs["maxlen"] == DLQ_MAXLEN
         assert kwargs["approximate"] is True
 
+    async def test_custom_dlq_maxlen_is_used(self):
+        redis = _redis_with_deliveries(times=6)
+        guard = ClaimDeadLetterGuard(
+            stream="events:clicks",
+            group="stats",
+            dlq_stream="events:clicks:dlq",
+            max_deliveries=5,
+            dlq_maxlen=100_000,
+        )
+        assert await guard.intercept(redis, "5-0", {}) is True
+        assert redis.xadd.await_args.kwargs["maxlen"] == 100_000
+
     async def test_bytes_delivery_metadata_is_handled(self):
         """The broker's internal client speaks bytes."""
         redis = AsyncMock()

workers/click_worker.py

@@ -351,6 +351,7 @@ def _register_group(
         group=group,
         dlq_stream=ce.dlq_stream,
         max_deliveries=ce.max_deliveries,
+        dlq_maxlen=ce.dlq_maxlen,
     )
 
     async def reader(body: Any) -> None:

workers/dlq.py

@@ -42,11 +42,13 @@ def __init__(
         group: str,
         dlq_stream: str,
         max_deliveries: int,
+        dlq_maxlen: int = DLQ_MAXLEN,
     ) -> None:
         self._stream = stream
         self._group = group
         self._dlq_stream = dlq_stream
         self._max_deliveries = max_deliveries
+        self._dlq_maxlen = dlq_maxlen
 
     async def intercept(self, redis: Any, message_id: str, payload: Any) -> bool:
         """Return True when the message was dead-lettered (skip processing).
@@ -68,7 +70,7 @@ async def intercept(self, redis: Any, message_id: str, payload: Any) -> bool:
                     DLQ_FIELD_GROUP: self._group,
                     DLQ_FIELD_REASON: _REASON_MAX_DELIVERIES,
                 },
-                maxlen=DLQ_MAXLEN,
+                maxlen=self._dlq_maxlen,
                 approximate=True,
             )
         except Exception as exc: