Feat/click event fanout - #213
Merged
Merged
Conversation
The bot-block decision for v1/emoji URLs moves ahead of the emit so the redirect response never depends on out-of-band processing.
…ading Module-level load_dotenv in the worker leaked real env values into the test process; it now runs only in the uvicorn factory entrypoint.
Contributor
Reviewer's GuideIntroduces a click-event fanout architecture: redirect routes now emit immutable ClickEvent objects to an abstract ClickEventSink, which can either process clicks inline (default) or enqueue them to a Redis Stream consumed by a new FastStream-based click worker with stats and hotness consumer groups; configuration, Docker wiring, Redis clients, and tests are updated accordingly while preserving existing redirect and analytics behavior by default. Sequence diagram for redirect handling with pluggable ClickEventSinksequenceDiagram
actor User
participant App as redirect_url
participant Sink as ClickEventSink
participant RedisQ as QueueRedis
participant Worker as StatsClickConsumer
User->>App: HTTP GET /{short_code}
App->>App: resolve URL, auth checks
App->>App: [bot_precheck for v1/emoji]
alt bot blocked
App-->>User: 403 ACCESS DENIED
else allowed
App->>App: build ClickEvent
alt CLICK_EVENTS_SINK=inline or fallback
App->>Sink: emit(event)
Sink->>Sink: InlineSink.emit
Sink->>App: track_click via ClickService
else CLICK_EVENTS_SINK=stream
App->>Sink: emit(event)
Sink->>RedisQ: XADD events:clicks
end
App-->>User: HTTP redirect 301/302
end
Note over RedisQ,Worker: Asynchronously, worker consumes stream
RedisQ-->>Worker: XREADGROUP click.recorded
Worker->>Worker: consume(payload)
Worker->>App: track_click via ClickService (same logic)
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
📝 WalkthroughWalkthroughIntroduces an optional click-event pipeline: a ChangesClick Event Pipeline
Estimated code review effort: 4 (Complex) | ~75 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant RedirectRoute
participant ClickSink
participant RedisStream
participant ClickWorker
Client->>RedirectRoute: GET short_code
RedirectRoute->>RedirectRoute: should_block_bot check
RedirectRoute->>ClickSink: emit(ClickEvent)
alt sink=stream
ClickSink->>RedisStream: XADD click event
else sink=inline or XADD fails
ClickSink->>ClickSink: fallback to inline track_click
end
RedirectRoute-->>Client: 302 redirect
RedisStream->>ClickWorker: XREADGROUP / XAUTOCLAIM
ClickWorker->>ClickWorker: consume via stats/hotness consumer
Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Zingzy
added
priority: high
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="docker-compose.prod.yml" line_range="239-240" />
<code_context>
+ --requirepass ${REDIS_QUEUE_PASSWORD}
+ volumes:
+ - redis-queue-data:/data
+ healthcheck:
+ test: ["CMD", "redis-cli", "ping"]
+ interval: 10s
+ timeout: 3s
</code_context>
<issue_to_address>
**issue (bug_risk):** Healthcheck will fail against password-protected Redis
Because Redis is started with `--requirepass ${REDIS_QUEUE_PASSWORD}`, `redis-cli ping` without `-a` will always return NOAUTH and mark the container unhealthy. Please update the healthcheck command to include the password (e.g. `redis-cli -a ${REDIS_QUEUE_PASSWORD} ping`) or another auth-aware check.
</issue_to_address>
### Comment 2
<location path="workers/click_worker.py" line_range="125-128" />
<code_context>
+ counter_redis=None,
+ )
+
+ if "stats" in groups:
+ geoip = GeoIPService(settings.geoip_country_db, settings.geoip_city_db)
+ url_cache = UrlCache(cache_redis, ttl_seconds=settings.redis.redis_ttl_seconds)
+ runtime.consumers["stats"] = StatsClickConsumer(
+ build_click_service(
+ ClickRepository(db["clicks"]),
</code_context>
<issue_to_address>
**suggestion (bug_risk):** GeoIPService lifecycle in the worker is not explicitly closed
`GeoIPService` is created here but not stored on `_WorkerRuntime` or closed on shutdown. If it holds file descriptors, threads, or MaxMind readers, this can leak resources over the worker’s lifetime or across restarts. Please retain it on `_WorkerRuntime` and close it in `aclose()`, consistent with the main app’s GeoIP teardown.
Suggested implementation:
```python
if "stats" in groups:
geoip = GeoIPService(settings.geoip_country_db, settings.geoip_city_db)
# Retain GeoIPService on the worker runtime so it can be closed on shutdown.
# This mirrors the main app's GeoIP lifecycle management.
runtime.geoip_service = geoip
url_cache = UrlCache(cache_redis, ttl_seconds=settings.redis.redis_ttl_seconds)
runtime.consumers["stats"] = StatsClickConsumer(
build_click_service(
ClickRepository(db["clicks"]),
UrlRepository(db["urlsV2"]),
LegacyUrlRepository(db["urls"]),
EmojiUrlRepository(db["emojis"]),
geoip,
url_cache,
)
)
```
To fully implement the suggestion, you should also:
1. In the `_WorkerRuntime` class (likely in `workers/click_worker.py` or a nearby module):
- Initialize the new attribute in `__init__` (or equivalent constructor) so it always exists:
```python
self.geoip_service = None
```
- Ensure any existing `geoip_service` passed in or set externally is assigned there if appropriate.
2. In `_WorkerRuntime.aclose()` (or whatever shutdown/cleanup coroutine is used):
- Add explicit teardown of the `GeoIPService`, following the same pattern as the main app:
```python
async def aclose(self) -> None:
# existing shutdown logic...
if getattr(self, "geoip_service", None) is not None:
# Use the correct close method based on GeoIPService's API.
# If it is async:
await self.geoip_service.aclose()
# or, if it is sync:
# self.geoip_service.close()
```
- If the main app uses a specific method name (e.g., `close()`, `aclose()`, `__aexit__`), mirror that here to keep behavior consistent.
3. If `_WorkerRuntime` is instantiated elsewhere with a `geoip_service` argument or similar, align the constructor signature and attribute name with the usage above (`geoip_service`) to avoid confusion and keep naming consistent across the codebase.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
There was a problem hiding this comment.
Pull request overview
This PR introduces a click-event fanout pipeline: redirect routes now emit immutable ClickEvent DTOs to a configurable sink (inline by default, Redis Stream when enabled), and a dedicated FastStream-based worker consumes the stream for analytics and optional hot-URL detection with DLQ handling.
Changes:
- Added
ClickEventDTO +ClickEventSinkabstraction with inline and Redis Stream sink implementations (with inline fallback on stream failures). - Added a FastStream Redis click worker with per-group reader/claimer subscribers and a claim-path DLQ guard for poison messages.
- Added click-event configuration (
ClickEventsSettings), Redis client improvements (labeled logging + richer options), docker-compose wiring for optional queue Redis + worker, and expanded unit/integration/smoke test coverage.
Reviewed changes
Copilot reviewed 36 out of 39 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| workers/dlq.py | Adds claim-path DLQ guard for poison messages in Redis Stream consumer groups. |
| workers/click_worker.py | Introduces FastStream-based click worker app factory with reader/claimer subscribers and health route. |
| workers/init.py | Package marker for worker modules. |
| uv.lock | Locks new dependency set including FastStream (Redis). |
| tests/unit/workers/test_dlq_guard.py | Unit tests for DLQ guard behavior and failure modes. |
| tests/unit/workers/test_click_worker_app.py | Unit tests for worker app factory wiring and subscriber registration. |
| tests/unit/test_click_events_settings.py | Tests for ClickEventsSettings defaults and env parsing/validation. |
| tests/unit/services/test_click_sinks.py | Tests for inline vs stream sink behavior and fallback semantics. |
| tests/unit/services/test_click_service.py | Updates tests for redirect_ms-based click context plumbing. |
| tests/unit/services/test_click_events.py | Tests for ClickEvent immutability and stream wire encoding/decoding. |
| tests/unit/services/test_click_consumers.py | Tests for stats consumer replay semantics and hotness detector behavior. |
| tests/smoke/test_click_sink_wiring.py | Smoke test validating sink wiring and inline fallback in app composition. |
| tests/integration/test_redirect.py | Updates redirect integration tests to override get_click_sink instead of click service. |
| tests/integration/test_redirect_routes.py | Expands redirect-route tests to assert emitted ClickEvent snapshots and bot-block behavior. |
| tests/integration/test_click_tracking.py | Updates click-tracking integration tests to validate event emission and error handling. |
| tests/conftest.py | Adds app.state.click_sink default mock in test lifespan. |
| services/click/sinks/stream.py | Implements Redis Stream sink with warning log + inline fallback on any XADD failure. |
| services/click/sinks/protocol.py | Adds sink protocol boundary for redirect → click processing. |
| services/click/sinks/inline.py | Implements inline sink that replays event into ClickService.track_click. |
| services/click/sinks/init.py | Exposes sink types from package. |
| services/click/service.py | Switches click context parameter from start_time to redirect_ms. |
| services/click/protocol.py | Updates ClickContext to carry redirect_ms rather than start_time. |
| services/click/handlers.py | Uses context.redirect_ms rather than recomputing from perf counter. |
| services/click/events.py | Adds ClickEvent DTO and Redis Stream wire helpers. |
| services/click/consumers/stats.py | Adds stats consumer that replays events via ClickService with drop vs retry semantics. |
| services/click/consumers/hotness.py | Adds hot-URL detection consumer using Redis window counters and pluggable actions. |
| services/click/consumers/init.py | Exports click consumer classes/types. |
| routes/redirect_routes.py | Refactors redirect route to emit ClickEvent through sink + pre-emit bot blocking for v1/emoji. |
| pyproject.toml | Adds faststream[redis] dependency. |
| infrastructure/cache/redis_client.py | Enhances Redis client creation (labels, keepalive, health checks, richer logs). |
| docker-compose.yml | Adds opt-in queue Redis + click-worker service behind click-events profile for local runs. |
| docker-compose.prod.yml | Adds production queue Redis + click-worker service configuration and health checks. |
| dependencies/wiring.py | Centralizes ClickService construction and wires click sink based on settings + queue Redis availability. |
| dependencies/services.py | Adds get_click_sink dependency and ClickSink annotated type. |
| dependencies/init.py | Re-exports new click sink dependency symbols. |
| config.py | Adds ClickEventsSettings and integrates it into AppSettings composition. |
| app.py | Connects optional queue Redis for stream mode and closes it on shutdown. |
| .env.example | Documents environment variables for enabling the click event pipeline. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (7)
docker-compose.prod.yml (1)
239-244: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winHealthcheck doesn't authenticate against the password-protected queue Redis.
redis-queueis started with--requirepass ${REDIS_QUEUE_PASSWORD}, but the healthcheck runs bareredis-cli ping. Redis returnsNOAUTH Authentication requiredfor this, andredis-cliexits with code0on that error — the container is reported "healthy" even if credentials are wrong/misconfigured, defeating the purpose of the check. This mirrors the existingspoo_redisservice's healthcheck but is reproduced here for the newredis-queueservice.🔒 Proposed fix — authenticate the healthcheck
healthcheck: - test: ["CMD", "redis-cli", "ping"] + test: ["CMD", "redis-cli", "-a", "${REDIS_QUEUE_PASSWORD}", "--no-auth-warning", "ping"] interval: 10s timeout: 3s retries: 5 start_period: 5s🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docker-compose.prod.yml` around lines 239 - 244, The healthcheck for redis-queue is currently using a plain redis-cli ping and does not authenticate, so it can report healthy even when the password is wrong. Update the redis-queue service’s healthcheck to pass the queue password when calling redis-cli, matching the authenticated pattern used by spoo_redis, and keep the fix localized to the healthcheck test block for that service.workers/dlq.py (1)
30-33: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUnused helper
_as_str.Not referenced anywhere in this file —
_times_deliveredonly usestimes_delivered(already handled viaint(times), Line 113), nevermessage_id/consumerbyte-decoding.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@workers/dlq.py` around lines 30 - 33, The helper _as_str in dlq.py is unused and should be removed. Delete the _as_str function and verify there are no remaining references in the DLQ module, since _times_delivered already handles its parsing directly with int(times) and does not need byte-decoding for message_id or consumer.workers/click_worker.py (2)
103-158: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winMongo/cache-Redis connections are opened even when unused.
mongo_clientandcache_redisare always created, but only used when"stats"is ingroups. A hotness-only worker (a scenario the module's own docstring anticipates — "a future deployment can split groups across containers", Lines 25-28) still pays for an unused Mongo pool and Redis connection.♻️ Proposed fix: build Mongo/cache-Redis only when stats is enabled
- ce = settings.click_events - mongo_client: AsyncMongoClient = AsyncMongoClient( - settings.db.mongodb_uri, - maxPoolSize=_WORKER_MONGO_MAX_POOL, - minPoolSize=1, - ) - db = mongo_client[settings.db.db_name] - - # Cache Redis is optional in the worker exactly as in the web app — - # without it the URL cache degrades to no-ops (max-clicks expiry just - # skips cache invalidation; resolve-side caching is the app's concern). - cache_redis = None - if settings.redis.redis_uri: - cache_redis = await create_redis_client(settings.redis.redis_uri, label="cache") - - runtime = _WorkerRuntime( - mongo_client=mongo_client, - cache_redis=cache_redis, - counter_redis=None, - ) - - if "stats" in groups: + ce = settings.click_events + mongo_client: AsyncMongoClient | None = None + cache_redis = None + + if "stats" in groups: + mongo_client = AsyncMongoClient( + settings.db.mongodb_uri, + maxPoolSize=_WORKER_MONGO_MAX_POOL, + minPoolSize=1, + ) + db = mongo_client[settings.db.db_name] + if settings.redis.redis_uri: + cache_redis = await create_redis_client(settings.redis.redis_uri, label="cache") + + runtime = _WorkerRuntime( + mongo_client=mongo_client, + cache_redis=cache_redis, + counter_redis=None, + ) + + if "stats" in groups:(
_WorkerRuntime.aclosewould also need aNoneguard formongo_client.)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@workers/click_worker.py` around lines 103 - 158, The `_build_runtime` setup eagerly opens `AsyncMongoClient` and the optional cache Redis even when only `"hotness"` is requested, so move those connection initializations behind the `"stats" in groups` branch and only pass them into `StatsClickConsumer` when needed. Update `_WorkerRuntime` so `mongo_client`/`cache_redis` can be absent in hotness-only runs, and add the corresponding `None` guard in `_WorkerRuntime.aclose` before closing the Mongo client.
239-247: 🩺 Stability & Availability | 🔵 TrivialStale consumer names accumulate in the Redis consumer group across restarts.
consumer_suffixis derived fromhostname-pid(Line 245), so every process restart registers a brand-new consumer name; old ones are never removed viaXGROUP DELCONSUMER. Functionally harmless (XAUTOCLAIM reclaims by idle time regardless of owner), butXINFO CONSUMERSwill grow unbounded over the worker's lifetime — worth a periodic cleanup job or accepting the metadata growth.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@workers/click_worker.py` around lines 239 - 247, The Redis consumer group setup in click_worker.consumer_for and the consumer_suffix registration currently creates a new consumer name on every restart, leaving stale entries behind. Fix this by adding a cleanup path that removes old consumers with XGROUP DELCONSUMER (or by reusing a stable consumer identity) when registering groups in _register_group / the group registration loop. Make sure the cleanup runs periodically or during startup before _register_group is called so XINFO CONSUMERS does not grow unbounded.app.py (1)
118-129: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider reusing
create_redis_clientfor the primary cache client too.The queue Redis client correctly adopts the new
create_redis_client(..., label=...)helper, but the primaryredis_clienta few lines above (94-116) still hand-rolls the identicalfrom_url+ping+ try/except logic. Now that the helper exists (and is used here), consolidating both call sites would remove duplication and give the cache client the same structuredlabel/error_typelogging.♻️ Proposed refactor
- redis_client = None - if settings.redis.redis_uri: - redis_client = aioredis.from_url( - settings.redis.redis_uri, - encoding="utf-8", - decode_responses=True, - socket_keepalive=True, - health_check_interval=30, - ) - try: - await redis_client.ping() - log.info( - "redis_connected", ttl_seconds=settings.redis.redis_ttl_seconds - ) - except Exception as e: - log.error("redis_connection_failed", error=str(e)) - redis_client = None - else: + redis_client = None + if settings.redis.redis_uri: + redis_client = await create_redis_client( + settings.redis.redis_uri, label="cache" + ) + else: log.warning( "redis_not_configured", detail="set REDIS_URI to enable caching" )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app.py` around lines 118 - 129, The queue Redis setup already uses create_redis_client with a label, but the primary cache client in app.py still duplicates the same from_url/ping/try-except flow. Refactor the cache-client initialization to call create_redis_client as well, so both the main redis_client and the click-events queue_redis share the same helper, structured label, and error_type logging.services/click/sinks/stream.py (1)
36-51: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueBroad
except Exceptionmasks non-Redis failures.Catching all exceptions around
xaddmeans bugs unrelated to Redis connectivity (e.g., a future serialization issue into_stream_fields) are silently routed through the fallback rather than surfaced distinctly. Consider narrowing toredis.exceptions.RedisError(and any connection-specific exceptions) so unexpected errors aren't masked by the "degrade gracefully" path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/click/sinks/stream.py` around lines 36 - 51, The emit method in ClickSink is catching too broadly and masking non-Redis bugs. Narrow the exception handling around the self._redis.xadd call in emit to Redis-specific failures (for example RedisError and any connection-related subclasses) so issues in to_stream_fields or other unexpected errors are not silently sent through _fallback.emit. Keep the fallback path for genuine Redis transport/write failures, and let non-Redis exceptions surface normally.services/click/events.py (1)
41-58: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueNested
UrlCacheDatais not itself frozen.
ClickEventusesConfigDict(frozen=True), but the embeddedurl: UrlCacheDatafield isn't guaranteed immutable at its own model level, so a consumer holding a reference could still mutateevent.urlin place, breaking the "immutable fact" guarantee documented at the top of the file.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@services/click/events.py` around lines 41 - 58, ClickEvent is marked frozen, but its nested url field can still be mutated if UrlCacheData itself is not frozen. Update the UrlCacheData model to be immutable as well, and confirm the ClickEvent model in events.py continues to use it as the embedded type so the full event remains an immutable fact.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@infrastructure/cache/redis_client.py`:
- Around line 35-49: The Redis connection handling in `from_url()` leaves the
created client/pool open when `ping()` fails, because the `RedisError` and
generic `Exception` branches return `None` without cleanup. Update those
exception paths in `redis_client.py` to close the constructed client by calling
`await client.aclose()` before returning, so any partially initialized Redis
client is always released after a failed `ping()`.
In `@services/click/events.py`:
- Around line 70-86: The malformed-event logging in from_stream_fields and the
matching click_event_from_payload path should not include slices of the raw
payload, because that can expose PII from ClickEvent data. Update the warning
logs in these branches to emit only structural diagnostics such as the
validation error, payload type, or field names, and keep any raw content fully
redacted or omitted. Use the existing from_stream_fields helper and the
analogous click_event_from_payload branch as the places to change so both JSON
and dict parsing paths behave consistently.
In `@services/click/sinks/stream.py`:
- Around line 36-43: The Redis client used by the click stream path is created
without a socket timeout, so `ClickStreamSink.emit()` can block indefinitely on
`xadd`. Update the Redis client configuration in `redis_client.py` to set a
bounded `socket_timeout` (and keep it appropriate for the redirect path), so
`emit()` can fail fast and continue with the inline fallback instead of hanging.
---
Nitpick comments:
In `@app.py`:
- Around line 118-129: The queue Redis setup already uses create_redis_client
with a label, but the primary cache client in app.py still duplicates the same
from_url/ping/try-except flow. Refactor the cache-client initialization to call
create_redis_client as well, so both the main redis_client and the click-events
queue_redis share the same helper, structured label, and error_type logging.
In `@docker-compose.prod.yml`:
- Around line 239-244: The healthcheck for redis-queue is currently using a
plain redis-cli ping and does not authenticate, so it can report healthy even
when the password is wrong. Update the redis-queue service’s healthcheck to pass
the queue password when calling redis-cli, matching the authenticated pattern
used by spoo_redis, and keep the fix localized to the healthcheck test block for
that service.
In `@services/click/events.py`:
- Around line 41-58: ClickEvent is marked frozen, but its nested url field can
still be mutated if UrlCacheData itself is not frozen. Update the UrlCacheData
model to be immutable as well, and confirm the ClickEvent model in events.py
continues to use it as the embedded type so the full event remains an immutable
fact.
In `@services/click/sinks/stream.py`:
- Around line 36-51: The emit method in ClickSink is catching too broadly and
masking non-Redis bugs. Narrow the exception handling around the
self._redis.xadd call in emit to Redis-specific failures (for example RedisError
and any connection-related subclasses) so issues in to_stream_fields or other
unexpected errors are not silently sent through _fallback.emit. Keep the
fallback path for genuine Redis transport/write failures, and let non-Redis
exceptions surface normally.
In `@workers/click_worker.py`:
- Around line 103-158: The `_build_runtime` setup eagerly opens
`AsyncMongoClient` and the optional cache Redis even when only `"hotness"` is
requested, so move those connection initializations behind the `"stats" in
groups` branch and only pass them into `StatsClickConsumer` when needed. Update
`_WorkerRuntime` so `mongo_client`/`cache_redis` can be absent in hotness-only
runs, and add the corresponding `None` guard in `_WorkerRuntime.aclose` before
closing the Mongo client.
- Around line 239-247: The Redis consumer group setup in
click_worker.consumer_for and the consumer_suffix registration currently creates
a new consumer name on every restart, leaving stale entries behind. Fix this by
adding a cleanup path that removes old consumers with XGROUP DELCONSUMER (or by
reusing a stable consumer identity) when registering groups in _register_group /
the group registration loop. Make sure the cleanup runs periodically or during
startup before _register_group is called so XINFO CONSUMERS does not grow
unbounded.
In `@workers/dlq.py`:
- Around line 30-33: The helper _as_str in dlq.py is unused and should be
removed. Delete the _as_str function and verify there are no remaining
references in the DLQ module, since _times_delivered already handles its parsing
directly with int(times) and does not need byte-decoding for message_id or
consumer.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 27893ebc-e566-4944-b722-f57fc6cb9ddf
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (38)
.env.exampleapp.pyconfig.pydependencies/__init__.pydependencies/services.pydependencies/wiring.pydocker-compose.prod.ymldocker-compose.ymlinfrastructure/cache/redis_client.pypyproject.tomlroutes/redirect_routes.pyservices/click/consumers/__init__.pyservices/click/consumers/hotness.pyservices/click/consumers/stats.pyservices/click/events.pyservices/click/handlers.pyservices/click/protocol.pyservices/click/service.pyservices/click/sinks/__init__.pyservices/click/sinks/inline.pyservices/click/sinks/protocol.pyservices/click/sinks/stream.pytests/conftest.pytests/integration/test_click_tracking.pytests/integration/test_redirect.pytests/integration/test_redirect_routes.pytests/smoke/test_click_sink_wiring.pytests/unit/services/test_click_consumers.pytests/unit/services/test_click_events.pytests/unit/services/test_click_service.pytests/unit/services/test_click_sinks.pytests/unit/test_click_events_settings.pytests/unit/workers/__init__.pytests/unit/workers/test_click_worker_app.pytests/unit/workers/test_dlq_guard.pyworkers/__init__.pyworkers/click_worker.pyworkers/dlq.py
… policy Replaces the client-side history sweeper: XADD ref_policy=ACKED sweeps fully-acknowledged entries past maxlen on every publish, server-side. Queue redis now requires >= 8.2 (gated at startup with inline fallback).
Password-hash sanitization moves into the event model (structural, not caller-remembered); the consumer protocol lives with its implementers; the queue redis gate is an extracted, tested module; the pre-emit bot decision is a named predicate; shared test factories replace per-file copies.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
infrastructure/queue_redis.py (1)
23-23: 🧹 Nitpick | 🔵 TrivialKnown Redis caveat with approximate
ACKEDtrimming worth confirming.Redis has an open bug where
XADD/XTRIM MAXLEN ~ ACKED(approximate trimming) can fail to evict entries even when fully acknowledged, first observed on 8.2.2 and reproduced on 8.4.0. Since this whole gate exists specifically to enableACKED-based trimming, it's worth confirming the deployed queue-Redis version/patch level isn't affected, or that exact trimming is used as a fallback if unacked backlog needs bounding.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@infrastructure/queue_redis.py` at line 23, The Redis version gate in MIN_QUEUE_REDIS_VERSION should account for the known approximate ACKED trimming bug on affected 8.2/8.4 patch levels. Update the version check logic in infrastructure/queue_redis.py to either require a safe fixed Redis patch level for the ACKED trimming path or add a fallback to exact trimming when the deployed version may be impacted. Refer to the queue Redis gating around MIN_QUEUE_REDIS_VERSION and the ACKED-based trimming feature so the compatibility check remains aligned with the actual Redis behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@workers/click_worker.py`:
- Around line 84-93: The telemetry shutdown in aclose cancels telemetry_tasks
but does not wait for them to finish, so they can still be running when
telemetry_redis is closed. Update aclose to cancel the tasks and then await
their completion (for example by gathering them and handling cancellation)
before closing telemetry_redis. Keep the existing close order for mongo_client,
cache_redis, counter_redis, and telemetry_redis, but ensure the telemetry_tasks
drain first.
---
Nitpick comments:
In `@infrastructure/queue_redis.py`:
- Line 23: The Redis version gate in MIN_QUEUE_REDIS_VERSION should account for
the known approximate ACKED trimming bug on affected 8.2/8.4 patch levels.
Update the version check logic in infrastructure/queue_redis.py to either
require a safe fixed Redis patch level for the ACKED trimming path or add a
fallback to exact trimming when the deployed version may be impacted. Refer to
the queue Redis gating around MIN_QUEUE_REDIS_VERSION and the ACKED-based
trimming feature so the compatibility check remains aligned with the actual
Redis behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 792346d0-d4b4-427f-97a2-28625d3aabac
📒 Files selected for processing (27)
app.pyconfig.pydocker-compose.prod.ymldocker-compose.ymlinfrastructure/queue_redis.pyk6/scenarios/_loadcmp.jsk6/scenarios/_realistic.jsroutes/redirect_routes.pyservices/click/bot_detection.pyservices/click/consumers/__init__.pyservices/click/consumers/protocol.pyservices/click/events.pyservices/click/sinks/stream.pytests/factories.pytests/integration/test_click_tracking.pytests/integration/test_redirect.pytests/integration/test_redirect_routes.pytests/unit/infrastructure/test_queue_redis.pytests/unit/services/test_bot_detection.pytests/unit/services/test_click_consumers.pytests/unit/services/test_click_events.pytests/unit/services/test_click_sinks.pytests/unit/test_click_events_settings.pytests/unit/workers/test_telemetry.pyworkers/click_worker.pyworkers/dlq.pyworkers/telemetry.py
🚧 Files skipped from review as they are similar to previous changes (15)
- services/click/consumers/init.py
- tests/unit/test_click_events_settings.py
- tests/unit/services/test_click_events.py
- services/click/sinks/stream.py
- docker-compose.yml
- tests/unit/services/test_click_sinks.py
- app.py
- docker-compose.prod.yml
- services/click/events.py
- tests/integration/test_click_tracking.py
- config.py
- tests/integration/test_redirect.py
- tests/unit/services/test_click_consumers.py
- tests/integration/test_redirect_routes.py
- workers/dlq.py
Zingzy
force-pushed
the
feat/click-event-fanout
branch
from
July 3, 2026 22:11
6d5c87e to
86a46d8
Compare
github-project-automation
Bot
moved this from 🏗️ In Progress
to ✔️ Done
in spoo.me Development Roadmap
There was a problem hiding this comment.
🧹 Nitpick comments (1)
docker-compose.prod.yml (1)
276-281: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valuePrefer the
redis-queueservice name over the hardcoded static IP.
CLICK_EVENTS_QUEUE_REDIS_URIuses172.30.0.45directly, while the rest of the file (e.g.,redis-exporter'sREDIS_ADDR=redis://redis:6379) resolves peers by service name over Docker's embedded DNS. Works today since the IP is statically pinned viaipv4_address, but ties this env var to network config that lives elsewhere in the file, risking silent breakage if the IP is ever changed without updating this string.♻️ Proposed change
- CLICK_EVENTS_QUEUE_REDIS_URI: redis://:${REDIS_QUEUE_PASSWORD}`@172.30.0.45`:6379/0 + CLICK_EVENTS_QUEUE_REDIS_URI: redis://:${REDIS_QUEUE_PASSWORD}`@redis-queue`:6379/0🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docker-compose.prod.yml` around lines 276 - 281, The CLICK_EVENTS_QUEUE_REDIS_URI value is hardcoded to a static IP, which should instead use the redis-queue service name like the other Docker service references in docker-compose.prod.yml. Update the environment entry in the relevant compose service to point at redis-queue via Docker DNS, using the existing CLICK_EVENTS_QUEUE_REDIS_URI setting and preserving the Redis password interpolation.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@docker-compose.prod.yml`:
- Around line 276-281: The CLICK_EVENTS_QUEUE_REDIS_URI value is hardcoded to a
static IP, which should instead use the redis-queue service name like the other
Docker service references in docker-compose.prod.yml. Update the environment
entry in the relevant compose service to point at redis-queue via Docker DNS,
using the existing CLICK_EVENTS_QUEUE_REDIS_URI setting and preserving the Redis
password interpolation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5315977c-10f6-4755-83d2-f798ff529397
📒 Files selected for processing (7)
docker-compose.prod.ymlinfrastructure/cache/redis_client.pyservices/click/events.pytests/unit/infrastructure/test_redis_client.pytests/unit/workers/test_dlq_guard.pyworkers/click_worker.pyworkers/dlq.py
🚧 Files skipped from review as they are similar to previous changes (4)
- infrastructure/cache/redis_client.py
- services/click/events.py
- tests/unit/workers/test_dlq_guard.py
- workers/click_worker.py
This was referenced Jul 4, 2026
This was referenced Jul 8, 2026
This was referenced Jul 18, 2026
Merged
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by Sourcery
Introduce a click event fanout pipeline that emits immutable ClickEvent objects from redirect routes into a Redis Stream, processes them asynchronously via a dedicated click worker, and falls back to inline tracking when the stream infrastructure is unavailable or misconfigured.
New Features:
Enhancements:
Build:
Deployment:
Tests:
Summary by CodeRabbit
New Features
Bug Fixes