feat(firehose): websocket firehose service - #200
Open
extreme4all wants to merge 10 commits into
Open
extreme4all wants to merge 10 commits into
extreme4all wants to merge 10 commits into
Conversation
added 10 commits
September 25, 2026 00:05
A raising value_deserializer kills aiokafka's internal fetcher task, which silently stalls the consumer on the first poison message. Invalid JSON now deserializes to a marker dict that fails model validation downstream, surfacing as a normal error-as-value to callers.
- ApiUserRepo.get_user_by_token: resolve an apiUser row by its token, used for stateless credential validation - has_permission: make the legacy token filter optional so permissions can be checked by user_id or user_name alone
The repository raises "voter_player_id or voter_player_name must be provided"; the test still expected the old wording.
- docker-compose: firehose dev service (bind-mounted sources for hot reload, graceful shutdown timeout) and RESET_TOPICS=false so topic resets stop breaking live consumers - mysql init: seed the firehose permission and document how ops allowlists a discord user (apiUser row + firehose permission)
Streams kafka topics to browsers and systems over websockets.
- GET /firehose: topic index; GET /firehose/{topic}: viewer page;
WS /firehose/{topic}: message stream (?anonymous=1 forces the shared
anonymous group)
- auth: discord access token (X-API-Key header or cookie) validated
against discord; the identity must exist as
apiUser.username = "discord_{id}" and hold the topic-specific
firehose.<topic> permission; otherwise anonymous; static dev keys via
settings keep the service usable without a database
- /login + /login/callback: discord oauth2 authorization code flow,
sets the credential cookie; /me returns identity + token
- consumer_manager: shared refcounted consumer per topic+group;
connection_manager: group-aware broadcast; poison messages and
transient consumer errors are skipped with backoff
- metrics: firehose_consumers/connections/messages_total/bytes_total
per topic and type, exposed on /metrics
- topics are a hardcoded catalog (topic -> event_queue model); only
players.scraped is exposed for now
- drop viewer.py and the html topics/login-callback pages from the base - api: json GET /firehose/topics, /login/callback redirects to /me - new _infra/_firehose_frontend: vite + react + tailwind manual test client, run locally via npm (FIREHOSE_URL picks the api target)
- compose-dev: add firehose service (dev target, hot reload, 8000:5000) - .dockerignore: exclude .git, node_modules, dist, caches, spec/test/development - root Dockerfile: uv 0.5.4 -> 0.5.13, fix broken uvx COPY
- ci: firehose.yml (push: lock/lint/format/tests, dispatch: build to quay; deploy job pending a firehose deployment.yaml) - compose: firehose service to target prod, drop bind mounts, healthcheck - projects/firehose: requires-python >=3.12,<3.13 to match containers - mypy: source-based resolution for bricks, scoped to the firehose base
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
1 participant
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
Adds a firehose service: streams kafka topics to browsers and systems over websockets.
New bricks:
bases/bot_detector/firehose+projects/firehose(compose service, port 8000). It consumesevent_queueand reuses theapiUser/permissions tables as a topic-scoped allowlist.The browser frontend lives in
_infra/_firehose_frontend(vite + react + tailwind) as a manual test/showcase client — not a brick, not a container; run it with npm.Endpoints (api only, no html)
GET /firehose/topicsWS /firehose/{topic}?anonymous=1forces the shared anonymous groupGET /login,/login/callback/meGET /meGET /metricsAuth (stateless)
X-API-Keyheader for systems, cookie for browsers) — validated against discord on every connect; no local tokens or sessionsapiUserrow withusername = "discord_{id}"and the topic-scopedfirehose.<topic>permission; otherwise anonymous; invalid tokens are rejected with 403Consumers & topics
(topic, group)(firehose-anonymous-<topic>latest /firehose-<topic>-<user>-<hash>earliest); connections of a group are broadcast totopic -> event_queuemodel) — onlyplayers.scrapedexposed for nowFrontend (manual test client)
_infra/_firehose_frontend— topic picker, token field, anonymous/connect/disconnect, pretty-printed message log. Same three access modes as the api (anonymous, discord login, pasted token). Vite dev server proxies/firehose,/login,/meto the api (FIREHOSE_URL, defaultlocalhost:5000); no CORS needed.Metrics
firehose_consumers,firehose_connections,firehose_messages_total,firehose_bytes_total— per topic andanonymous|keyed, on/metrics.Related changes
fix(event_queue): kafka value deserializer never raises (a poison message used to kill aiokafka's fetcher and stall the consumer silently)feat(database):ApiUserRepo.get_user_by_token+ optional-tokenhas_permission(permission check by user id)chore(infra): firehose compose service (dev hot reload) andfirehosepermission seed;RESET_TOPICS=falserefactor(firehose): embedded html pages removed from the base; frontend extracted to_infra/_firehose_frontend;/firehose/topicsis json;/login/callbackredirects to/mechore(infra): firehose api service in compose-dev;.dockerignorecovers.git/node_modules/caches; root Dockerfile uv pinned to0.5.13chore(firehose): ci workflow (firehose.yml: push = lock/lint/format/tests, dispatch = build to quay; deploy job pending a deployment.yaml); compose firehose image istarget: prod+ healthcheck;requires-python >=3.12,<3.13so the local venv matches containers; mypy configured and green for the firehose baseTest plan
/loginflow with a real discord account,/metricscounters while streamingnpm run buildgreen; flows verified through the vite proxy against the compose-dev api (topics json, page 200)bd/firehose:prdbuilds; compose files validate; mypy/ruff/pytest green locally