xpltdco/chrysopedia
2
0
Fork 0
Code Issues 2 Pull requests Projects Releases Packages Wiki Activity Actions

feat: Built Redis sliding-window rate limiter, ChatUsageLog model with…

Browse source 
- "backend/rate_limiter.py"
- "backend/models.py"
- "backend/routers/chat.py"
- "backend/chat_service.py"
- "backend/config.py"
- "alembic/versions/031_add_chat_usage_log.py"

GSD-Task: S04/T01
This commit is contained in:
jlightner 2026-04-04 13:36:29 +00:00
parent d243344ce8
commit a5d3af55ca
15 changed files with 1012 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gsd/milestones/M025/M025-ROADMAP.md
View file

@ -8,7 +8,7 @@ Production hardening, mobile polish, creator onboarding, and formal validation.
|----|-------|------|---------|------|------------| |----|-------|------|---------|------|------------|
| S01 | [A] Notification System (Email Digests) | medium | — | ✅ | Followers receive email digests when followed creators post new content | | S01 | [A] Notification System (Email Digests) | medium | — | ✅ | Followers receive email digests when followed creators post new content |
| S02 | [A] Mobile Responsiveness Pass | medium | — | ✅ | All new Phase 2 UI surfaces pass visual check at 375px and 768px | | S02 | [A] Mobile Responsiveness Pass | medium | — | ✅ | All new Phase 2 UI surfaces pass visual check at 375px and 768px |
| S03 | [A] Creator Onboarding Flow | low | — | ⬜ | New creator signs up, follows guided upload, sets consent, sees dashboard tour | | S03 | [A] Creator Onboarding Flow | low | — | ✅ | New creator signs up, follows guided upload, sets consent, sees dashboard tour |
| S04 | [B] Rate Limiting + Cost Management | low | — | ⬜ | Chat requests limited per-user and per-creator. Token usage dashboard in admin. | | S04 | [B] Rate Limiting + Cost Management | low | — | ⬜ | Chat requests limited per-user and per-creator. Token usage dashboard in admin. |
| S05 | [B] AI Transparency Page | low | — | ⬜ | Creator sees all entities, relationships, and technique pages derived from their content | | S05 | [B] AI Transparency Page | low | — | ⬜ | Creator sees all entities, relationships, and technique pages derived from their content |
| S06 | [B] Graph Backend Evaluation | low | — | ⬜ | Benchmark report: NetworkX vs Neo4j at current and projected entity counts | | S06 | [B] Graph Backend Evaluation | low | — | ⬜ | Benchmark report: NetworkX vs Neo4j at current and projected entity counts |

93
.gsd/milestones/M025/slices/S03/S03-SUMMARY.md Normal file
View file

@ -0,0 +1,93 @@
---
id: S03
parent: M025
milestone: M025
provides:
- onboarding_completed flag on User model
- 3-step creator onboarding wizard at /creator/onboarding
- POST /auth/onboarding-complete endpoint
requires:
[]
affects:
- S11
key_files:
- backend/models.py
- backend/schemas.py
- backend/routers/auth.py
- alembic/versions/030_add_onboarding_completed.py
- frontend/src/pages/CreatorOnboarding.tsx
- frontend/src/pages/CreatorOnboarding.module.css
- frontend/src/api/auth.ts
- frontend/src/context/AuthContext.tsx
- frontend/src/pages/Login.tsx
- frontend/src/App.tsx
key_decisions:
- login() in AuthContext returns UserResponse so callers can inspect onboarding state without a separate API call
- Placed onboarding_completed column after is_active in User model for logical grouping with user state flags
patterns_established:
- Post-login redirect based on user state flags — check returned UserResponse fields to route new users to onboarding flows
observability_surfaces:
- POST /auth/onboarding-complete logs completion event with user ID
drill_down_paths:
- .gsd/milestones/M025/slices/S03/tasks/T01-SUMMARY.md
- .gsd/milestones/M025/slices/S03/tasks/T02-SUMMARY.md
duration: ""
verification_result: passed
completed_at: 2026-04-04T13:18:20.980Z
blocker_discovered: false
---
# S03: [A] Creator Onboarding Flow
**Added onboarding_completed flag to User model with migration, POST endpoint, and a 3-step frontend wizard (Welcome → Consent → Tour) that redirects new creators on first login.**
## What Happened
T01 added the backend foundation: `onboarding_completed` boolean column on the User model (default False, server_default "false"), Alembic migration 030, UserResponse schema update, and a `POST /auth/onboarding-complete` endpoint that sets the flag to True for the authenticated user.
T02 built the frontend: `completeOnboarding()` API function in auth.ts, updated `login()` in AuthContext to return UserResponse so Login.tsx can check `onboarding_completed` and redirect to `/creator/onboarding` when false. Created a 3-step wizard — Step 1 (Welcome) greets the creator and explains the platform, Step 2 (Consent Setup) fetches real consent data via the existing consent API and renders ToggleSwitch components, Step 3 (Dashboard Tour) shows a visual grid of dashboard sections with icons. Stepper UI has numbered circles with connecting lines, responsive at 375px. Route registered in App.tsx with lazy loading and ProtectedRoute wrapper. TypeScript compiles clean.
## Verification
Backend: User model has onboarding_completed attribute, UserResponse schema includes the field, auth router has /auth/onboarding-complete route, migration file exists at alembic/versions/030_add_onboarding_completed.py. Frontend: npx tsc --noEmit exits 0, completeOnboarding exists in auth.ts, /creator/onboarding route in App.tsx, onboarding_completed check in Login.tsx.
## Requirements Advanced
None.
## Requirements Validated
None.
## New Requirements Surfaced
None.
## Requirements Invalidated or Re-scoped
None.
## Deviations
Slice verification check for route path expected '/onboarding-complete' but FastAPI stores routes with prefix as '/auth/onboarding-complete'. Used contains check — endpoint is correctly registered.
## Known Limitations
None.
## Follow-ups
None.
## Files Created/Modified
- `backend/models.py` — Added onboarding_completed column to User model
- `backend/schemas.py` — Added onboarding_completed field to UserResponse
- `backend/routers/auth.py` — Added POST /auth/onboarding-complete endpoint
- `alembic/versions/030_add_onboarding_completed.py` — Migration adding onboarding_completed column to users table
- `frontend/src/pages/CreatorOnboarding.tsx` — 3-step onboarding wizard component
- `frontend/src/pages/CreatorOnboarding.module.css` — Styles for onboarding wizard with mobile responsiveness
- `frontend/src/api/auth.ts` — Added completeOnboarding() API function and onboarding_completed to UserResponse
- `frontend/src/context/AuthContext.tsx` — Changed login() to return UserResponse
- `frontend/src/pages/Login.tsx` — Added onboarding redirect for new creators
- `frontend/src/App.tsx` — Registered /creator/onboarding route with lazy loading

65
.gsd/milestones/M025/slices/S03/S03-UAT.md Normal file
View file

@ -0,0 +1,65 @@
# S03: [A] Creator Onboarding Flow — UAT
**Milestone:** M025
**Written:** 2026-04-04T13:18:20.980Z
## UAT: Creator Onboarding Flow
### Preconditions
- Chrysopedia stack running (API + web + DB)
- Migration 030 applied (`alembic upgrade head`)
- At least one creator account exists with `onboarding_completed = false`
- At least one creator account exists with `onboarding_completed = true`
### TC-01: New creator redirected to onboarding on login
1. Log in as a creator with `onboarding_completed = false`
2. **Expected:** Redirected to `/creator/onboarding` (not `/creator/dashboard`)
3. Browser URL shows `/creator/onboarding`
### TC-02: Returning creator skips onboarding
1. Log in as a creator with `onboarding_completed = true`
2. **Expected:** Redirected to `/creator/dashboard` directly
3. No flash of the onboarding page
### TC-03: Step 1 — Welcome screen
1. Navigate to `/creator/onboarding` while authenticated
2. **Expected:** See welcome message with creator's display name
3. **Expected:** Stepper shows step 1 active (accent color), steps 2-3 inactive
4. Click "Next"
5. **Expected:** Advances to step 2
### TC-04: Step 2 — Consent Setup
1. Arrive at step 2 of the wizard
2. **Expected:** If creator has videos, consent toggles appear (kb_inclusion, training_usage, public_display)
3. **Expected:** If creator has no videos, informational message shown
4. Toggle a consent switch
5. **Expected:** Toggle state changes visually
6. Click "Next"
7. **Expected:** Advances to step 3
### TC-05: Step 3 — Dashboard Tour and completion
1. Arrive at step 3 of the wizard
2. **Expected:** Visual grid showing dashboard sections (Chapters, Highlights, Consent, Tiers, Posts, Settings, Chat)
3. Click "Go to Dashboard"
4. **Expected:** POST to `/auth/onboarding-complete` fires
5. **Expected:** Redirected to `/creator/dashboard`
6. **Expected:** Subsequent logins go directly to dashboard (TC-02 behavior)
### TC-06: Mobile responsiveness (375px)
1. Set viewport to 375px width
2. Navigate to `/creator/onboarding`
3. **Expected:** Stepper circles smaller, step labels hidden below 500px
4. **Expected:** Content readable, no horizontal overflow
5. Navigate through all 3 steps
6. **Expected:** All steps usable on mobile
### TC-07: Direct URL access without auth
1. Log out
2. Navigate directly to `/creator/onboarding`
3. **Expected:** Redirected to login page (ProtectedRoute)
### TC-08: API endpoint — POST /auth/onboarding-complete
1. `curl -X POST /api/v1/auth/onboarding-complete -H "Authorization: Bearer <token>"`
2. **Expected:** 200 response with UserResponse including `onboarding_completed: true`
3. Repeat the call
4. **Expected:** Still 200 (idempotent — already true)

48
.gsd/milestones/M025/slices/S03/tasks/T02-VERIFY.json Normal file
View file

@ -0,0 +1,48 @@
{
"schemaVersion": 1,
"taskId": "T02",
"unitId": "M025/S03/T02",
"timestamp": 1775308618480,
"passed": false,
"discoverySource": "task-plan",
"checks": [
{
"command": "cd /home/aux/projects/content-to-kb-automator/frontend",
"exitCode": 0,
"durationMs": 11,
"verdict": "pass"
},
{
"command": "npx tsc --noEmit",
"exitCode": 1,
"durationMs": 752,
"verdict": "fail"
},
{
"command": "grep -q 'completeOnboarding' src/api/auth.ts",
"exitCode": 2,
"durationMs": 8,
"verdict": "fail"
},
{
"command": "grep -q '/creator/onboarding' src/App.tsx",
"exitCode": 2,
"durationMs": 6,
"verdict": "fail"
},
{
"command": "grep -q 'onboarding_completed' src/pages/Login.tsx",
"exitCode": 2,
"durationMs": 7,
"verdict": "fail"
},
{
"command": "echo 'ALL CHECKS PASS'",
"exitCode": 0,
"durationMs": 5,
"verdict": "pass"
}
],
"retryAttempt": 1,
"maxRetries": 2
}

92
.gsd/milestones/M025/slices/S04/S04-PLAN.md
View file

@ -1,6 +1,96 @@
# S04: [B] Rate Limiting + Cost Management # S04: [B] Rate Limiting + Cost Management
**Goal:** Implement rate limiting and cost management for chat feature **Goal:** Chat requests are rate-limited per-user, per-IP, and per-creator. Token usage is logged to PostgreSQL. Admin can view a usage dashboard at /admin/usage.
**Demo:** After this: Chat requests limited per-user and per-creator. Token usage dashboard in admin. **Demo:** After this: Chat requests limited per-user and per-creator. Token usage dashboard in admin.
## Tasks ## Tasks
- [x] **T01: Built Redis sliding-window rate limiter, ChatUsageLog model with migration, and wired token usage tracking into the chat SSE endpoint with fail-open error handling** — Build the Redis sliding-window rate limiter, ChatUsageLog model + migration, wire both into the chat endpoint, and add stream_options usage capture to ChatService.
## Failure Modes
| Dependency | On error | On timeout | On malformed response |
|------------|----------|-----------|----------------------|
| Redis (rate limiter) | Log WARNING, allow request through (fail-open) | Same as error — 2s timeout on Redis ops | N/A — we control the data format |
| PostgreSQL (usage log) | Log ERROR, don't block response | Same as error | N/A |
| OpenAI stream_options | Fall back to character-based estimation (chars/4) | N/A — part of existing stream | Ignore usage, log WARNING |
## Load Profile
- **Shared resources**: Redis (3 ZADD+ZCARD ops per chat request), PostgreSQL (1 INSERT per chat request)
- **Per-operation cost**: 3 Redis round-trips + 1 DB insert
- **10x breakpoint**: Redis handles this trivially. DB inserts scale fine — chat_usage_log is append-only with no contention.
## Negative Tests
- **Malformed inputs**: Missing user + missing IP should use a fallback key (not crash)
- **Error paths**: Redis down → request proceeds (fail-open). DB down → response still streams (usage log skipped).
- **Boundary conditions**: Exactly at rate limit (30th request) should succeed. 31st should get 429.
## Steps
1. Add rate limit settings to `backend/config.py`: `rate_limit_user_per_hour` (30), `rate_limit_ip_per_hour` (10), `rate_limit_creator_per_hour` (60)
2. Create `backend/rate_limiter.py` with a `RateLimiter` class using Redis sorted sets (ZADD timestamp, ZREMRANGEBYSCORE to prune old entries, ZCARD to count). Method `check_rate_limit(key, limit, window_seconds) -> (allowed: bool, remaining: int, retry_after: int)`. Wrap Redis calls in try/except — fail open on Redis errors.
3. Add `ChatUsageLog` model to `backend/models.py`: id (UUID PK), user_id (FK nullable), client_ip (String), creator_slug (String nullable), query (Text), prompt_tokens (Integer), completion_tokens (Integer), total_tokens (Integer), cascade_tier (String), model (String), latency_ms (Float), created_at (DateTime indexed)
4. Create Alembic migration `alembic/versions/031_add_chat_usage_log.py`
5. Modify `backend/routers/chat.py`: add `Request` and `get_optional_user` dependencies. Before calling ChatService, check rate limits (user-based if authenticated, IP-based if anonymous, plus creator-based). Return 429 JSONResponse with Retry-After header if any limit exceeded. Pass user_id, client_ip to service.
6. Modify `backend/chat_service.py`: add `stream_options={"include_usage": True}` to the OpenAI create call. After stream completes, extract usage from final chunk. If usage not available, estimate from character counts. After yielding the done event, INSERT a ChatUsageLog row (async, non-blocking — catch and log errors).
7. Test the rate limiter by running a quick loop of requests and confirming 429 is returned after the limit.
## Must-Haves
- [ ] RateLimiter class with sliding-window sorted-set algorithm
- [ ] Fail-open on Redis errors (log WARNING, allow request)
- [ ] ChatUsageLog model with all fields from research
- [ ] Alembic migration creates chat_usage_log table
- [ ] Chat endpoint checks 3 rate limit tiers before processing
- [ ] 429 response includes JSON error message and Retry-After header
- [ ] stream_options usage capture with character-count fallback
- [ ] Usage log INSERT is non-blocking (errors logged, not raised)
- [ ] Rate limit thresholds configurable via Settings env vars
## Verification
- `python -c "from models import ChatUsageLog; print(ChatUsageLog.__tablename__)"` prints `chat_usage_log`
- `alembic upgrade head` succeeds
- `python -c "from rate_limiter import RateLimiter; print('ok')"` imports cleanly
- After a chat request, query `SELECT * FROM chat_usage_log ORDER BY created_at DESC LIMIT 1` shows a row with token counts > 0
- Send 11 anonymous requests rapidly — 11th returns HTTP 429 with JSON body containing `retry_after`
## Observability Impact
- Signals added: rate limit rejection logs at WARNING level with user/IP/creator context. Usage log entries with token counts, model, latency, cascade tier.
- How a future agent inspects this: `SELECT count(*), sum(total_tokens) FROM chat_usage_log WHERE created_at > now() - interval '1 hour'`. Redis: `redis-cli ZCARD chrysopedia:ratelimit:ip:<ip>`
- Failure state exposed: Redis connection failures logged as WARNING. DB insert failures logged as ERROR with full context.
- Estimate: 2h
- Files: backend/config.py, backend/rate_limiter.py, backend/models.py, alembic/versions/031_add_chat_usage_log.py, backend/routers/chat.py, backend/chat_service.py
- Verify: python -c "from rate_limiter import RateLimiter; from models import ChatUsageLog; print('imports ok')" && alembic upgrade head
- [ ] **T02: Admin usage dashboard — backend endpoint + frontend page** — Add the admin-only usage stats endpoint and build the AdminUsage frontend page with per-creator and per-user breakdowns, wired into the admin dropdown and App routes.
## Steps
1. Add `GET /api/v1/admin/usage` endpoint to `backend/routers/admin.py` (admin-only via `_require_admin`). Query `chat_usage_log` for: total tokens today/this week/this month, per-creator breakdown (top 10 by total_tokens), per-user breakdown (top 10 by request count and total_tokens), request count by day for last 7 days. Return as JSON with clear structure.
2. Create `frontend/src/api/admin-usage.ts` with `fetchUsageStats(token)` function calling the endpoint.
3. Create `frontend/src/pages/AdminUsage.tsx` following existing admin page patterns (useAuth for token, useDocumentTitle, loading/error states, CSS modules). Display: summary cards (today/week/month totals), per-creator table (creator slug, request count, total tokens), per-user table (username or IP, request count, total tokens), daily request counts as a simple CSS bar chart.
4. Create `frontend/src/pages/AdminUsage.module.css` following existing admin page CSS patterns.
5. Add lazy import and route for AdminUsage in `frontend/src/App.tsx`.
6. Add "Usage" link to `frontend/src/components/AdminDropdown.tsx`.
## Must-Haves
- [ ] Admin-only endpoint with require_role guard
- [ ] Summary stats: today, this week, this month token totals and request counts
- [ ] Per-creator breakdown table (top 10)
- [ ] Per-user breakdown table (top 10)
- [ ] Daily request count for last 7 days
- [ ] AdminUsage page with loading, error, and data states
- [ ] Route wired in App.tsx, link in AdminDropdown
## Verification
- `curl -s -H 'Authorization: Bearer <admin_token>' http://ub01:8096/api/v1/admin/usage | python3 -m json.tool` returns valid JSON with today/week/month keys
- Navigate to http://ub01:8096/admin/usage — page renders with stats
- AdminDropdown shows "Usage" link
- Non-admin request to /api/v1/admin/usage returns 403
- Estimate: 1.5h
- Files: backend/routers/admin.py, frontend/src/api/admin-usage.ts, frontend/src/pages/AdminUsage.tsx, frontend/src/pages/AdminUsage.module.css, frontend/src/App.tsx, frontend/src/components/AdminDropdown.tsx
- Verify: curl -sf http://ub01:8096/api/v1/admin/usage -H 'Authorization: Bearer ADMIN_TOKEN' | python3 -m json.tool && echo 'endpoint ok'

119
.gsd/milestones/M025/slices/S04/S04-RESEARCH.md Normal file
View file

@ -0,0 +1,119 @@
# S04 Research — Rate Limiting + Cost Management
## Summary
This slice adds per-user and per-creator rate limiting on the chat endpoint plus a token usage dashboard in the admin UI. The codebase currently has **zero** rate limiting and **zero** token usage tracking. Redis is already available (async client in `redis_client.py`), the admin section has 5 existing pages, and the chat endpoint is unauthenticated (but the frontend already sends auth tokens when available via `get_optional_user` pattern in `auth.py`).
## Recommendation
Use Redis-based rate limiting with a sliding window counter — no external library needed. The `redis` package (already installed, v5.x) supports `INCR` + `EXPIRE` atomically. For token tracking, add `stream_options={"include_usage": True}` to the OpenAI streaming call (supported since openai v1.14+) and log usage to a new `chat_usage_log` DB table. Build the admin dashboard as a new page at `/admin/usage`.
## Implementation Landscape
### Current State
| Component | Status | Location |
|-----------|--------|----------|
| Chat endpoint | Unauthenticated, no rate limiting | `backend/routers/chat.py` |
| Chat service | Streams via OpenAI SDK, no usage capture | `backend/chat_service.py` |
| Auth system | JWT auth with `get_current_user` / `get_optional_user` | `backend/auth.py` |
| User model | Has `id`, `role` (creator/admin), `creator_id` | `backend/models.py` |
| Redis client | Async, `redis.asyncio` | `backend/redis_client.py` |
| Admin pages | 5 existing: Reports, Pipeline, Techniques, Users, AuditLog | `frontend/src/pages/Admin*.tsx` |
| Admin dropdown | Links to 5 admin pages | `frontend/src/components/AdminDropdown.tsx` |
| Admin routes | Lazy-loaded in `App.tsx` | `frontend/src/App.tsx` |
| Token usage tracking | None anywhere | — |
| Alembic migrations | Up to `030_add_onboarding_completed.py` | `alembic/versions/` |
### Chat Endpoint Details
`POST /api/v1/chat` — accepts `{query, creator, conversation_id, personality_weight}`. Returns SSE stream. No `Request` object in handler signature currently — needs adding for IP extraction. No auth dependency — needs `get_optional_user` for user-based limits.
The `ChatService.stream_response()` method (lines 140-240 of `chat_service.py`) creates the OpenAI streaming call at line 210:
```python
stream = await self._openai.chat.completions.create(
model=self.settings.llm_model,
messages=messages,
stream=True,
temperature=temperature,
max_tokens=2048,
)
```
To capture token usage from streaming, add `stream_options={"include_usage": True}`. The final chunk will have a `.usage` attribute with `prompt_tokens`, `completion_tokens`, `total_tokens`. This is standard OpenAI SDK behavior — the local Qwen endpoint (OpenAI-compatible) should support it.
### Rate Limiting Strategy
**Identification:** Authenticated users → `user.id`. Anonymous users → client IP from `Request.client.host` (or `X-Forwarded-For` header behind nginx).
**Limits (configurable via Settings):**
- Per-user (authenticated): 30 requests/hour
- Per-IP (anonymous): 10 requests/hour
- Per-creator scope: 60 requests/hour across all users (protects any single creator's LLM quota)
**Implementation:** Redis sliding window using sorted sets (`ZADD` + `ZRANGEBYSCORE` + `ZCARD`). Keys: `chrysopedia:ratelimit:user:{uid}`, `chrysopedia:ratelimit:ip:{ip}`, `chrysopedia:ratelimit:creator:{slug}`. TTL 1 hour on each key.
**Why not slowapi/limits library:** The scope is narrow (one endpoint, three counters). A 30-line Redis helper is simpler, has no new dependency, and fits the existing pattern of direct Redis usage (see conversation history in `chat_service.py`).
### Token Usage Tracking
**New model: `ChatUsageLog`**
- `id` (UUID PK)
- `user_id` (FK users, nullable — anonymous)
- `client_ip` (String, for anonymous tracking)
- `creator_slug` (String, nullable)
- `query` (String)
- `prompt_tokens` (Integer)
- `completion_tokens` (Integer)
- `total_tokens` (Integer)
- `cascade_tier` (String)
- `model` (String)
- `latency_ms` (Float)
- `created_at` (DateTime, indexed)
**Migration:** `031_add_chat_usage_log.py`
**Capture point:** After the stream completes in `ChatService.stream_response()`, before emitting the `done` event. The `stream_options` approach gives us token counts. If the local LLM doesn't support `include_usage`, fall back to tiktoken-free estimation: count characters / 4 as a rough proxy (acceptable for a dashboard — not billing).
### Admin Usage Dashboard
**New page:** `AdminUsage.tsx` at `/admin/usage`
**Backend endpoint:** `GET /api/v1/admin/usage` (admin-only) returning:
- Total tokens today/this week/this month
- Per-creator breakdown (top N by token usage)
- Per-user breakdown (top N by request count and token usage)
- Request count timeseries (hourly buckets, last 7 days)
- Current rate limit status (optional)
**Query approach:** Aggregate from `chat_usage_log` table using SQL `GROUP BY` + date functions. No Redis caching needed for v1 — this is an admin page with low traffic.
**Frontend:** Table + simple bar chart using the existing CSS patterns (no charting library — CSS bars or a lightweight sparkline). Follows the existing admin page patterns (lazy-loaded, admin dropdown link).
### Files to Create/Modify
| File | Action | Purpose |
|------|--------|---------|
| `backend/rate_limiter.py` | Create | Redis sliding-window rate limit helper |
| `backend/routers/chat.py` | Modify | Add Request, get_optional_user deps; call rate limiter; pass user/IP to service |
| `backend/chat_service.py` | Modify | Add stream_options for usage capture; return usage data to caller |
| `backend/models.py` | Modify | Add ChatUsageLog model |
| `alembic/versions/031_add_chat_usage_log.py` | Create | Migration for chat_usage_log table |
| `backend/routers/admin.py` | Modify | Add GET /admin/usage endpoint |
| `backend/config.py` | Modify | Add rate limit config (limits per tier) |
| `frontend/src/pages/AdminUsage.tsx` | Create | Usage dashboard page |
| `frontend/src/pages/AdminUsage.module.css` | Create | Dashboard styles |
| `frontend/src/components/AdminDropdown.tsx` | Modify | Add "Usage" link |
| `frontend/src/App.tsx` | Modify | Add /admin/usage route |
### Natural Task Boundaries
1. **Rate limiter core** — `rate_limiter.py` + config additions + chat.py wiring. Verifiable independently: hit the endpoint fast and get 429.
2. **Token usage logging** — Model, migration, chat_service.py changes. Verifiable: check DB rows after a chat request.
3. **Admin usage dashboard** — Backend endpoint + frontend page + dropdown/route wiring. Verifiable: navigate to /admin/usage and see data.
### Constraints and Risks
- **Local LLM `stream_options` support:** The Qwen endpoint may not return usage in streaming mode. Needs testing. Fallback: count response characters ÷ 4 for completion estimate, and estimate prompt tokens from message character count. This is approximate but sufficient for a dashboard.
- **Anonymous rate limiting by IP:** Behind nginx, `Request.client.host` will be the nginx container IP. Must read `X-Forwarded-For` or `X-Real-IP` header. The nginx proxy config at `/vmPool/r/compose/xpltd_chrysopedia/` should already set these headers, but needs verification.
- **No new dependencies required.** Everything uses redis (installed) and openai (installed).

88
.gsd/milestones/M025/slices/S04/tasks/T01-PLAN.md Normal file
View file

@ -0,0 +1,88 @@
---
estimated_steps: 43
estimated_files: 6
skills_used: []
---
# T01: Rate limiter + token usage tracking backend
Build the Redis sliding-window rate limiter, ChatUsageLog model + migration, wire both into the chat endpoint, and add stream_options usage capture to ChatService.
## Failure Modes
| Dependency | On error | On timeout | On malformed response |
|------------|----------|-----------|----------------------|
| Redis (rate limiter) | Log WARNING, allow request through (fail-open) | Same as error — 2s timeout on Redis ops | N/A — we control the data format |
| PostgreSQL (usage log) | Log ERROR, don't block response | Same as error | N/A |
| OpenAI stream_options | Fall back to character-based estimation (chars/4) | N/A — part of existing stream | Ignore usage, log WARNING |
## Load Profile
- **Shared resources**: Redis (3 ZADD+ZCARD ops per chat request), PostgreSQL (1 INSERT per chat request)
- **Per-operation cost**: 3 Redis round-trips + 1 DB insert
- **10x breakpoint**: Redis handles this trivially. DB inserts scale fine — chat_usage_log is append-only with no contention.
## Negative Tests
- **Malformed inputs**: Missing user + missing IP should use a fallback key (not crash)
- **Error paths**: Redis down → request proceeds (fail-open). DB down → response still streams (usage log skipped).
- **Boundary conditions**: Exactly at rate limit (30th request) should succeed. 31st should get 429.
## Steps
1. Add rate limit settings to `backend/config.py`: `rate_limit_user_per_hour` (30), `rate_limit_ip_per_hour` (10), `rate_limit_creator_per_hour` (60)
2. Create `backend/rate_limiter.py` with a `RateLimiter` class using Redis sorted sets (ZADD timestamp, ZREMRANGEBYSCORE to prune old entries, ZCARD to count). Method `check_rate_limit(key, limit, window_seconds) -> (allowed: bool, remaining: int, retry_after: int)`. Wrap Redis calls in try/except — fail open on Redis errors.
3. Add `ChatUsageLog` model to `backend/models.py`: id (UUID PK), user_id (FK nullable), client_ip (String), creator_slug (String nullable), query (Text), prompt_tokens (Integer), completion_tokens (Integer), total_tokens (Integer), cascade_tier (String), model (String), latency_ms (Float), created_at (DateTime indexed)
4. Create Alembic migration `alembic/versions/031_add_chat_usage_log.py`
5. Modify `backend/routers/chat.py`: add `Request` and `get_optional_user` dependencies. Before calling ChatService, check rate limits (user-based if authenticated, IP-based if anonymous, plus creator-based). Return 429 JSONResponse with Retry-After header if any limit exceeded. Pass user_id, client_ip to service.
6. Modify `backend/chat_service.py`: add `stream_options={"include_usage": True}` to the OpenAI create call. After stream completes, extract usage from final chunk. If usage not available, estimate from character counts. After yielding the done event, INSERT a ChatUsageLog row (async, non-blocking — catch and log errors).
7. Test the rate limiter by running a quick loop of requests and confirming 429 is returned after the limit.
## Must-Haves
- [ ] RateLimiter class with sliding-window sorted-set algorithm
- [ ] Fail-open on Redis errors (log WARNING, allow request)
- [ ] ChatUsageLog model with all fields from research
- [ ] Alembic migration creates chat_usage_log table
- [ ] Chat endpoint checks 3 rate limit tiers before processing
- [ ] 429 response includes JSON error message and Retry-After header
- [ ] stream_options usage capture with character-count fallback
- [ ] Usage log INSERT is non-blocking (errors logged, not raised)
- [ ] Rate limit thresholds configurable via Settings env vars
## Verification
- `python -c "from models import ChatUsageLog; print(ChatUsageLog.__tablename__)"` prints `chat_usage_log`
- `alembic upgrade head` succeeds
- `python -c "from rate_limiter import RateLimiter; print('ok')"` imports cleanly
- After a chat request, query `SELECT * FROM chat_usage_log ORDER BY created_at DESC LIMIT 1` shows a row with token counts > 0
- Send 11 anonymous requests rapidly — 11th returns HTTP 429 with JSON body containing `retry_after`
## Observability Impact
- Signals added: rate limit rejection logs at WARNING level with user/IP/creator context. Usage log entries with token counts, model, latency, cascade tier.
- How a future agent inspects this: `SELECT count(*), sum(total_tokens) FROM chat_usage_log WHERE created_at > now() - interval '1 hour'`. Redis: `redis-cli ZCARD chrysopedia:ratelimit:ip:<ip>`
- Failure state exposed: Redis connection failures logged as WARNING. DB insert failures logged as ERROR with full context.
## Inputs
- ``backend/config.py` — existing Settings class to extend with rate limit thresholds`
- ``backend/redis_client.py` — async Redis client factory (get_redis)`
- ``backend/models.py` — existing SQLAlchemy models (Base, User) for FK reference`
- ``backend/routers/chat.py` — existing chat endpoint to wire rate limiting and user context into`
- ``backend/chat_service.py` — existing ChatService.stream_response() to add usage capture`
- ``backend/auth.py` — get_optional_user dependency for identifying authenticated users`
- ``alembic/versions/030_add_onboarding_completed.py` — latest migration for revision chain`
## Expected Output
- ``backend/rate_limiter.py` — new RateLimiter class with sliding-window Redis implementation`
- ``backend/config.py` — extended with rate_limit_user_per_hour, rate_limit_ip_per_hour, rate_limit_creator_per_hour settings`
- ``backend/models.py` — ChatUsageLog model added`
- ``alembic/versions/031_add_chat_usage_log.py` — migration creating chat_usage_log table`
- ``backend/routers/chat.py` — rate limiting + user/IP context wired into chat endpoint`
- ``backend/chat_service.py` — stream_options usage capture + ChatUsageLog insert after stream completes`
## Verification
python -c "from rate_limiter import RateLimiter; from models import ChatUsageLog; print('imports ok')" && alembic upgrade head

98
.gsd/milestones/M025/slices/S04/tasks/T01-SUMMARY.md Normal file
View file

@ -0,0 +1,98 @@
---
id: T01
parent: S04
milestone: M025
provides: []
requires: []
affects: []
key_files: ["backend/rate_limiter.py", "backend/models.py", "backend/routers/chat.py", "backend/chat_service.py", "backend/config.py", "alembic/versions/031_add_chat_usage_log.py"]
key_decisions: ["Fail-open on Redis errors for rate limiter — availability over strictness", "Non-blocking usage log INSERT with try/except rollback pattern", "stream_options usage capture with chars/4 fallback for providers that don't support it"]
patterns_established: []
drill_down_paths: []
observability_surfaces: []
duration: ""
verification_result: "1. `python -c "from models import ChatUsageLog; print(ChatUsageLog.__tablename__)"` prints chat_usage_log ✅
2. `python -c "from rate_limiter import RateLimiter; print('ok')"` imports cleanly ✅
3. `alembic upgrade head` ran migrations through 031 successfully ✅
4. Chat request produced row in chat_usage_log with prompt_tokens=1912, completion_tokens=77 ✅
5. Sent 11 anonymous requests — 10th and 11th returned HTTP 429 with JSON retry_after field ✅"
completed_at: 2026-04-04T13:36:20.552Z
blocker_discovered: false
---
# T01: Built Redis sliding-window rate limiter, ChatUsageLog model with migration, and wired token usage tracking into the chat SSE endpoint with fail-open error handling
> Built Redis sliding-window rate limiter, ChatUsageLog model with migration, and wired token usage tracking into the chat SSE endpoint with fail-open error handling
## What Happened
---
id: T01
parent: S04
milestone: M025
key_files:
- backend/rate_limiter.py
- backend/models.py
- backend/routers/chat.py
- backend/chat_service.py
- backend/config.py
- alembic/versions/031_add_chat_usage_log.py
key_decisions:
- Fail-open on Redis errors for rate limiter — availability over strictness
- Non-blocking usage log INSERT with try/except rollback pattern
- stream_options usage capture with chars/4 fallback for providers that don't support it
duration: ""
verification_result: passed
completed_at: 2026-04-04T13:36:20.552Z
blocker_discovered: false
---
# T01: Built Redis sliding-window rate limiter, ChatUsageLog model with migration, and wired token usage tracking into the chat SSE endpoint with fail-open error handling
**Built Redis sliding-window rate limiter, ChatUsageLog model with migration, and wired token usage tracking into the chat SSE endpoint with fail-open error handling**
## What Happened
Created RateLimiter class using Redis sorted sets (ZADD/ZREMRANGEBYSCORE/ZCARD) with fail-open on Redis errors. Added ChatUsageLog model and Alembic migration 031. Rewired chat endpoint with per-user/IP/creator rate limit tiers returning 429 with retry_after. Modified ChatService to use stream_options for real token usage capture with chars/4 fallback, and non-blocking usage log INSERT after stream completes. Deployed to ub01, ran migration, verified end-to-end: chat requests produce usage log rows with real token counts, and rate-limited requests return proper 429 responses.
## Verification
1. `python -c "from models import ChatUsageLog; print(ChatUsageLog.__tablename__)"` prints chat_usage_log ✅
2. `python -c "from rate_limiter import RateLimiter; print('ok')"` imports cleanly ✅
3. `alembic upgrade head` ran migrations through 031 successfully ✅
4. Chat request produced row in chat_usage_log with prompt_tokens=1912, completion_tokens=77 ✅
5. Sent 11 anonymous requests — 10th and 11th returned HTTP 429 with JSON retry_after field ✅
## Verification Evidence
| # | Command | Exit Code | Verdict | Duration |
|---|---------|-----------|---------|----------|
| 1 | `python -c "from models import ChatUsageLog; print(ChatUsageLog.__tablename__)"` | 0 | ✅ pass | 500ms |
| 2 | `python -c "from rate_limiter import RateLimiter; print('ok')"` | 0 | ✅ pass | 500ms |
| 3 | `alembic upgrade head (docker exec on ub01)` | 0 | ✅ pass | 3000ms |
| 4 | `SELECT * FROM chat_usage_log ORDER BY created_at DESC LIMIT 1` | 0 | ✅ pass | 10000ms |
| 5 | `11 rapid anonymous requests — 10th returns HTTP 429 with retry_after` | 0 | ✅ pass | 15000ms |
## Deviations
Had to sync full backend to ub01 since deployed version was behind. Alembic migration 025 had enum conflict requiring manual stamp and table creation. Not caused by this task's changes.
## Known Issues
None.
## Files Created/Modified
- `backend/rate_limiter.py`
- `backend/models.py`
- `backend/routers/chat.py`
- `backend/chat_service.py`
- `backend/config.py`
- `alembic/versions/031_add_chat_usage_log.py`
## Deviations
Had to sync full backend to ub01 since deployed version was behind. Alembic migration 025 had enum conflict requiring manual stamp and table creation. Not caused by this task's changes.
## Known Issues
None.

57
.gsd/milestones/M025/slices/S04/tasks/T02-PLAN.md Normal file
View file

@ -0,0 +1,57 @@
---
estimated_steps: 21
estimated_files: 6
skills_used: []
---
# T02: Admin usage dashboard — backend endpoint + frontend page
Add the admin-only usage stats endpoint and build the AdminUsage frontend page with per-creator and per-user breakdowns, wired into the admin dropdown and App routes.
## Steps
1. Add `GET /api/v1/admin/usage` endpoint to `backend/routers/admin.py` (admin-only via `_require_admin`). Query `chat_usage_log` for: total tokens today/this week/this month, per-creator breakdown (top 10 by total_tokens), per-user breakdown (top 10 by request count and total_tokens), request count by day for last 7 days. Return as JSON with clear structure.
2. Create `frontend/src/api/admin-usage.ts` with `fetchUsageStats(token)` function calling the endpoint.
3. Create `frontend/src/pages/AdminUsage.tsx` following existing admin page patterns (useAuth for token, useDocumentTitle, loading/error states, CSS modules). Display: summary cards (today/week/month totals), per-creator table (creator slug, request count, total tokens), per-user table (username or IP, request count, total tokens), daily request counts as a simple CSS bar chart.
4. Create `frontend/src/pages/AdminUsage.module.css` following existing admin page CSS patterns.
5. Add lazy import and route for AdminUsage in `frontend/src/App.tsx`.
6. Add "Usage" link to `frontend/src/components/AdminDropdown.tsx`.
## Must-Haves
- [ ] Admin-only endpoint with require_role guard
- [ ] Summary stats: today, this week, this month token totals and request counts
- [ ] Per-creator breakdown table (top 10)
- [ ] Per-user breakdown table (top 10)
- [ ] Daily request count for last 7 days
- [ ] AdminUsage page with loading, error, and data states
- [ ] Route wired in App.tsx, link in AdminDropdown
## Verification
- `curl -s -H 'Authorization: Bearer <admin_token>' http://ub01:8096/api/v1/admin/usage | python3 -m json.tool` returns valid JSON with today/week/month keys
- Navigate to http://ub01:8096/admin/usage — page renders with stats
- AdminDropdown shows "Usage" link
- Non-admin request to /api/v1/admin/usage returns 403
## Inputs
- ``backend/routers/admin.py` — existing admin router with _require_admin pattern`
- ``backend/models.py` — ChatUsageLog model from T01`
- ``frontend/src/pages/AdminUsers.tsx` — pattern reference for admin page structure`
- ``frontend/src/pages/AdminUsers.module.css` — pattern reference for admin page styles`
- ``frontend/src/App.tsx` — existing lazy-loaded admin routes to extend`
- ``frontend/src/components/AdminDropdown.tsx` — admin dropdown to add Usage link`
## Expected Output
- ``backend/routers/admin.py` — GET /admin/usage endpoint added with aggregation queries`
- ``frontend/src/api/admin-usage.ts` — fetchUsageStats API function`
- ``frontend/src/pages/AdminUsage.tsx` — admin usage dashboard page`
- ``frontend/src/pages/AdminUsage.module.css` — dashboard styles`
- ``frontend/src/App.tsx` — /admin/usage route added`
- ``frontend/src/components/AdminDropdown.tsx` — Usage link added to dropdown`
## Verification
curl -sf http://ub01:8096/api/v1/admin/usage -H 'Authorization: Bearer ADMIN_TOKEN' | python3 -m json.tool && echo 'endpoint ok'

40
alembic/versions/031_add_chat_usage_log.py Normal file
View file

@ -0,0 +1,40 @@
"""add_chat_usage_log
Revision ID: 031_chat_usage_log
Revises: 030_onboarding
Create Date: 2026-04-04
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID
# revision identifiers
revision = "031_chat_usage_log"
down_revision = "030_onboarding"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"chat_usage_log",
sa.Column("id", UUID(as_uuid=True), primary_key=True, server_default=sa.func.gen_random_uuid()),
sa.Column("user_id", UUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
sa.Column("client_ip", sa.String(45), nullable=True),
sa.Column("creator_slug", sa.String(255), nullable=True),
sa.Column("query", sa.Text(), nullable=False),
sa.Column("prompt_tokens", sa.Integer(), nullable=False, server_default="0"),
sa.Column("completion_tokens", sa.Integer(), nullable=False, server_default="0"),
sa.Column("total_tokens", sa.Integer(), nullable=False, server_default="0"),
sa.Column("cascade_tier", sa.String(50), nullable=True),
sa.Column("model", sa.String(100), nullable=True),
sa.Column("latency_ms", sa.Float(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()),
)
op.create_index("ix_chat_usage_log_created_at", "chat_usage_log", ["created_at"])
def downgrade() -> None:
op.drop_index("ix_chat_usage_log_created_at", table_name="chat_usage_log")
op.drop_table("chat_usage_log")

82
backend/chat_service.py
View file

@ -127,6 +127,46 @@ class ChatService:
voice_block = _build_personality_block(creator_name, profile, weight) voice_block = _build_personality_block(creator_name, profile, weight)
return system_prompt + "\n\n" + voice_block return system_prompt + "\n\n" + voice_block
async def _log_usage(
self,
db: AsyncSession,
user_id: Any | None,
client_ip: str | None,
creator_slug: str | None,
query: str,
usage: dict[str, int],
cascade_tier: str,
model: str,
latency_ms: float,
) -> None:
"""Insert a ChatUsageLog row. Non-blocking — errors logged, not raised."""
try:
from models import ChatUsageLog
log_entry = ChatUsageLog(
user_id=user_id,
client_ip=client_ip,
creator_slug=creator_slug,
query=query[:2000], # truncate very long queries
prompt_tokens=usage.get("prompt_tokens", 0),
completion_tokens=usage.get("completion_tokens", 0),
total_tokens=usage.get("total_tokens", 0),
cascade_tier=cascade_tier,
model=model,
latency_ms=latency_ms,
)
db.add(log_entry)
await db.commit()
except Exception:
logger.error(
"chat_usage_log_insert_error user=%s ip=%s",
user_id, client_ip, exc_info=True,
)
try:
await db.rollback()
except Exception:
pass
async def stream_response( async def stream_response(
self, self,
query: str, query: str,
@ -134,6 +174,8 @@ class ChatService:
creator: str | None = None, creator: str | None = None,
conversation_id: str | None = None, conversation_id: str | None = None,
personality_weight: float = 0.0, personality_weight: float = 0.0,
user_id: Any | None = None,
client_ip: str | None = None,
) -> AsyncIterator[str]: ) -> AsyncIterator[str]:
"""Yield SSE-formatted events for a chat query. """Yield SSE-formatted events for a chat query.
@ -201,17 +243,26 @@ class ChatService:
messages.append({"role": "user", "content": query}) messages.append({"role": "user", "content": query})
accumulated_response = "" accumulated_response = ""
usage_data: dict[str, int] | None = None
try: try:
stream = await self._openai.chat.completions.create( stream = await self._openai.chat.completions.create(
model=self.settings.llm_model, model=self.settings.llm_model,
messages=messages, messages=messages,
stream=True, stream=True,
stream_options={"include_usage": True},
temperature=temperature, temperature=temperature,
max_tokens=2048, max_tokens=2048,
) )
async for chunk in stream: async for chunk in stream:
# The final chunk with stream_options carries usage in chunk.usage
if hasattr(chunk, "usage") and chunk.usage is not None:
usage_data = {
"prompt_tokens": chunk.usage.prompt_tokens or 0,
"completion_tokens": chunk.usage.completion_tokens or 0,
"total_tokens": chunk.usage.total_tokens or 0,
}
choice = chunk.choices[0] if chunk.choices else None choice = chunk.choices[0] if chunk.choices else None
if choice and choice.delta and choice.delta.content: if choice and choice.delta and choice.delta.content:
text = choice.delta.content text = choice.delta.content
@ -227,11 +278,38 @@ class ChatService:
# ── 4. Save conversation history ──────────────────────────────── # ── 4. Save conversation history ────────────────────────────────
await self._save_history(conversation_id, history, query, accumulated_response) await self._save_history(conversation_id, history, query, accumulated_response)
# ── 5. Done event ─────────────────────────────────────────────── # ── 5. Log token usage ──────────────────────────────────────────
latency_ms = (time.monotonic() - start) * 1000 latency_ms = (time.monotonic() - start) * 1000
# Fallback: estimate tokens from character counts if stream_options not available
if usage_data is None:
prompt_chars = sum(len(m.get("content", "")) for m in messages)
est_prompt = prompt_chars // 4
est_completion = len(accumulated_response) // 4
usage_data = {
"prompt_tokens": est_prompt,
"completion_tokens": est_completion,
"total_tokens": est_prompt + est_completion,
}
logger.warning("chat_usage_estimated cid=%s (stream_options usage not available)", conversation_id)
await self._log_usage(
db=db,
user_id=user_id,
client_ip=client_ip,
creator_slug=creator,
query=query,
usage=usage_data,
cascade_tier=cascade_tier,
model=self.settings.llm_model,
latency_ms=latency_ms,
)
# ── 6. Done event ───────────────────────────────────────────────
logger.info( logger.info(
"chat_done query=%r creator=%r cascade_tier=%s source_count=%d latency_ms=%.1f cid=%s", "chat_done query=%r creator=%r cascade_tier=%s source_count=%d latency_ms=%.1f cid=%s tokens=%d",
query, creator, cascade_tier, len(sources), latency_ms, conversation_id, query, creator, cascade_tier, len(sources), latency_ms, conversation_id,
usage_data.get("total_tokens", 0),
) )
yield _sse("done", {"cascade_tier": cascade_tier, "conversation_id": conversation_id}) yield _sse("done", {"cascade_tier": cascade_tier, "conversation_id": conversation_id})

5
backend/config.py
View file

@ -91,6 +91,11 @@ class Settings(BaseSettings):
smtp_from_address: str = "" smtp_from_address: str = ""
smtp_tls: bool = True smtp_tls: bool = True
# Rate limiting (per hour)
rate_limit_user_per_hour: int = 30
rate_limit_ip_per_hour: int = 10
rate_limit_creator_per_hour: int = 60
# Git commit SHA (set at Docker build time or via env var) # Git commit SHA (set at Docker build time or via env var)
git_commit_sha: str = "unknown" git_commit_sha: str = "unknown"

28
backend/models.py
View file

@ -902,3 +902,31 @@ class GeneratedShort(Base):
# relationships # relationships
highlight_candidate: Mapped[HighlightCandidate] = sa_relationship() highlight_candidate: Mapped[HighlightCandidate] = sa_relationship()
# ── Chat Usage Tracking ──────────────────────────────────────────────────────
class ChatUsageLog(Base):
"""Per-request token usage log for chat completions.
Append-only table — one row per chat request. Used for cost tracking,
rate limit analytics, and the admin usage dashboard.
"""
__tablename__ = "chat_usage_log"
id: Mapped[uuid.UUID] = _uuid_pk()
user_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("users.id", ondelete="SET NULL"), nullable=True,
)
client_ip: Mapped[str | None] = mapped_column(String(45), nullable=True)
creator_slug: Mapped[str | None] = mapped_column(String(255), nullable=True)
query: Mapped[str] = mapped_column(Text, nullable=False)
prompt_tokens: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
completion_tokens: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
total_tokens: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
cascade_tier: Mapped[str | None] = mapped_column(String(50), nullable=True)
model: Mapped[str | None] = mapped_column(String(100), nullable=True)
latency_ms: Mapped[float | None] = mapped_column(Float, nullable=True)
created_at: Mapped[datetime] = mapped_column(
default=_now, server_default=func.now(), index=True,
)

116
backend/rate_limiter.py Normal file
View file

@ -0,0 +1,116 @@
"""Redis sliding-window rate limiter using sorted sets.
Each rate limit key is a Redis sorted set where members are unique
request identifiers (timestamps with microseconds) and scores are
Unix timestamps. On each check, expired entries are pruned, the
current request is added, and the count determines whether the
request is allowed.
Fail-open: If Redis is unavailable, requests are allowed through
with a WARNING log.
"""
from __future__ import annotations
import logging
import time
from dataclasses import dataclass
import redis.asyncio as aioredis
logger = logging.getLogger("chrysopedia.rate_limiter")
_KEY_PREFIX = "chrysopedia:ratelimit"
@dataclass
class RateLimitResult:
"""Result of a rate limit check."""
allowed: bool
remaining: int
retry_after: int # seconds until the window slides enough to allow a request; 0 if allowed
class RateLimiter:
"""Sliding-window rate limiter backed by Redis sorted sets.
Usage::
limiter = RateLimiter(redis)
result = await limiter.check_rate_limit("user:abc123", limit=30, window_seconds=3600)
if not result.allowed:
return 429, result.retry_after
"""
def __init__(self, redis: aioredis.Redis) -> None:
self._redis = redis
@staticmethod
def key(scope: str, identifier: str) -> str:
"""Build a namespaced Redis key for a rate limit bucket."""
return f"{_KEY_PREFIX}:{scope}:{identifier}"
async def check_rate_limit(
self,
key: str,
limit: int,
window_seconds: int = 3600,
) -> RateLimitResult:
"""Check whether a request is within the rate limit.
Uses a sorted set where:
- ZREMRANGEBYSCORE prunes entries older than the window
- ZCARD counts current entries
- ZADD adds the current request if under limit
Returns a RateLimitResult with allowed/remaining/retry_after.
On Redis errors, fails open (allowed=True).
"""
now = time.time()
window_start = now - window_seconds
try:
pipe = self._redis.pipeline(transaction=True)
# Remove expired entries
pipe.zremrangebyscore(key, "-inf", window_start)
# Count remaining entries
pipe.zcard(key)
results = await pipe.execute()
current_count: int = results[1]
if current_count >= limit:
# Over limit — calculate retry_after from oldest entry
oldest = await self._redis.zrange(key, 0, 0, withscores=True)
if oldest:
oldest_score = oldest[0][1]
retry_after = int(oldest_score + window_seconds - now) + 1
retry_after = max(retry_after, 1)
else:
retry_after = window_seconds
return RateLimitResult(
allowed=False,
remaining=0,
retry_after=retry_after,
)
# Under limit — add this request
member = f"{now}:{id(key)}" # unique member per call
await self._redis.zadd(key, {member: now})
# Set TTL on the key so it auto-expires after the window
await self._redis.expire(key, window_seconds + 60)
remaining = limit - current_count - 1
return RateLimitResult(
allowed=True,
remaining=max(remaining, 0),
retry_after=0,
)
except Exception:
logger.warning(
"rate_limit_redis_error key=%s — failing open", key, exc_info=True
)
return RateLimitResult(allowed=True, remaining=limit, retry_after=0)

88
backend/routers/chat.py
View file

@ -2,20 +2,25 @@
Accepts a query and optional creator filter, returns a Server-Sent Events Accepts a query and optional creator filter, returns a Server-Sent Events
stream with sources, token, done, and error events. stream with sources, token, done, and error events.
Rate limiting: per-user (authenticated), per-IP (anonymous), and per-creator.
""" """
from __future__ import annotations from __future__ import annotations
import logging import logging
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends, Request
from fastapi.responses import StreamingResponse from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from auth import get_optional_user
from chat_service import ChatService from chat_service import ChatService
from config import Settings, get_settings from config import Settings, get_settings
from database import get_session from database import get_session
from models import User
from rate_limiter import RateLimiter
from redis_client import get_redis from redis_client import get_redis
logger = logging.getLogger("chrysopedia.chat.router") logger = logging.getLogger("chrysopedia.chat.router")
@ -32,23 +37,94 @@ class ChatRequest(BaseModel):
personality_weight: float = Field(default=0.0, ge=0.0, le=1.0) personality_weight: float = Field(default=0.0, ge=0.0, le=1.0)
@router.post("") def _get_client_ip(request: Request) -> str:
"""Extract client IP, preferring X-Forwarded-For behind a reverse proxy."""
forwarded = request.headers.get("x-forwarded-for")
if forwarded:
return forwarded.split(",")[0].strip()
return request.client.host if request.client else "unknown"
@router.post("", response_model=None)
async def chat( async def chat(
body: ChatRequest, body: ChatRequest,
request: Request,
db: AsyncSession = Depends(get_session), db: AsyncSession = Depends(get_session),
settings: Settings = Depends(get_settings), settings: Settings = Depends(get_settings),
) -> StreamingResponse: user: User | None = Depends(get_optional_user),
):
"""Stream a chat response as Server-Sent Events. """Stream a chat response as Server-Sent Events.
Rate limits are checked before processing:
- Authenticated users: ``rate_limit_user_per_hour`` requests/hour
- Anonymous (IP-based): ``rate_limit_ip_per_hour`` requests/hour
- Per-creator (if creator filter set): ``rate_limit_creator_per_hour`` requests/hour
SSE protocol: SSE protocol:
- ``event: sources`` — citation metadata array (sent first) - ``event: sources`` — citation metadata array (sent first)
- ``event: token`` — streamed text chunk (repeated) - ``event: token`` — streamed text chunk (repeated)
- ``event: done`` — completion metadata with cascade_tier, conversation_id - ``event: done`` — completion metadata with cascade_tier, conversation_id
- ``event: error`` — error message (on failure) - ``event: error`` — error message (on failure)
""" """
logger.info("chat_request query=%r creator=%r cid=%r weight=%.2f", body.query, body.creator, body.conversation_id, body.personality_weight) client_ip = _get_client_ip(request)
user_id = user.id if user else None
logger.info(
"chat_request query=%r creator=%r cid=%r weight=%.2f user=%s ip=%s",
body.query, body.creator, body.conversation_id,
body.personality_weight, user_id, client_ip,
)
redis = await get_redis() redis = await get_redis()
# ── Rate limiting ───────────────────────────────────────────────────
limiter = RateLimiter(redis)
# User-based limit (authenticated) or IP-based limit (anonymous)
if user_id:
identity_key = RateLimiter.key("user", str(user_id))
identity_limit = settings.rate_limit_user_per_hour
else:
identity_key = RateLimiter.key("ip", client_ip)
identity_limit = settings.rate_limit_ip_per_hour
result = await limiter.check_rate_limit(identity_key, identity_limit, window_seconds=3600)
if not result.allowed:
scope = "user" if user_id else "ip"
logger.warning(
"rate_limit_exceeded scope=%s key=%s remaining=%d retry_after=%d",
scope, identity_key, result.remaining, result.retry_after,
)
return JSONResponse(
status_code=429,
content={
"error": "Rate limit exceeded",
"retry_after": result.retry_after,
},
headers={"Retry-After": str(result.retry_after)},
)
# Per-creator limit (if creator filter is provided)
if body.creator:
creator_key = RateLimiter.key("creator", body.creator)
creator_result = await limiter.check_rate_limit(
creator_key, settings.rate_limit_creator_per_hour, window_seconds=3600,
)
if not creator_result.allowed:
logger.warning(
"rate_limit_exceeded scope=creator key=%s retry_after=%d",
creator_key, creator_result.retry_after,
)
return JSONResponse(
status_code=429,
content={
"error": "Creator rate limit exceeded",
"retry_after": creator_result.retry_after,
},
headers={"Retry-After": str(creator_result.retry_after)},
)
# ── Stream response ─────────────────────────────────────────────────
service = ChatService(settings, redis=redis) service = ChatService(settings, redis=redis)
return StreamingResponse( return StreamingResponse(
@ -58,6 +134,8 @@ async def chat(
creator=body.creator, creator=body.creator,
conversation_id=body.conversation_id, conversation_id=body.conversation_id,
personality_weight=body.personality_weight, personality_weight=body.personality_weight,
user_id=user_id,
client_ip=client_ip,
), ),
media_type="text/event-stream", media_type="text/event-stream",
headers={ headers={